Coverage for little_loops / cli / loop / info.py: 4%
858 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-15 17:30 -0500
1"""ll-loop info subcommands: list, history, show."""
3from __future__ import annotations
5import argparse
6import os
7from datetime import datetime
8from pathlib import Path
9from typing import Any
11from little_loops.cli.loop._helpers import (
12 get_builtin_loops_dir,
13 load_loop_with_spec,
14 resolve_loop_path,
15)
16from little_loops.cli.loop.diagram_modes import resolve_facets
17from little_loops.cli.loop.layout import ( # noqa: F401
18 _EDGE_LABEL_COLORS,
19 _box_inner_lines,
20 _colorize_diagram_labels,
21 _colorize_label,
22 _render_fsm_diagram,
23)
24from little_loops.cli.output import colorize, print_json, strip_ansi, terminal_width
25from little_loops.fsm import is_runnable_loop
26from little_loops.fsm.fragments import resolve_inheritance
27from little_loops.fsm.schema import FSMLoop, StateConfig
28from little_loops.fsm.validation import load_and_validate
29from little_loops.logger import Logger
32def _load_loop_meta(path: Path) -> dict[str, Any]:
33 """Return metadata from a loop YAML file (description, category, labels)."""
34 import yaml
36 try:
37 with open(path) as f:
38 spec = yaml.safe_load(f) or {}
39 try:
40 spec = resolve_inheritance(spec, path.parent)
41 except Exception:
42 pass
43 desc_raw = spec.get("description", "") or ""
44 if desc_raw.strip():
45 raw_lines = desc_raw.splitlines()
46 desc = raw_lines[0]
47 if len(raw_lines) > 1:
48 desc += "…"
49 else:
50 desc = ""
51 category = spec.get("category", "") or ""
52 labels: list[str] = spec.get("labels", []) or []
53 visibility = spec.get("visibility", "public") or "public"
54 if visibility not in ("public", "internal", "example"):
55 visibility = "public"
56 return {
57 "description": desc,
58 "category": category,
59 "labels": labels,
60 "visibility": visibility,
61 }
62 except Exception:
63 return {"description": "", "category": "", "labels": [], "visibility": "public"}
66def cmd_list(
67 args: argparse.Namespace,
68 loops_dir: Path,
69) -> int:
70 """List loops."""
71 status_filter = getattr(args, "status", None)
72 if getattr(args, "running", False) or status_filter:
73 from little_loops.fsm.persistence import list_running_loops
75 states = list_running_loops(loops_dir)
76 if status_filter:
77 states = [s for s in states if s.status == status_filter]
78 if not states:
79 if status_filter:
80 print(f"No loops with status: {status_filter}")
81 return 1
82 print("No running loops")
83 return 0
84 if getattr(args, "json", False):
85 print_json([s.to_dict() for s in states])
86 return 0
87 print(colorize("Running loops:", "1"))
88 _STATUS_COLORS = {"running": "32", "interrupted": "33", "stopped": "2", "starting": "33"}
90 # Group by loop_name to avoid duplicate rows for multi-instance loops
91 from collections import defaultdict
93 groups: dict[str, list] = defaultdict(list)
94 for state in states:
95 groups[state.loop_name].append(state)
97 for loop_name_key, group_states in groups.items():
98 if len(group_states) == 1:
99 state = group_states[0]
100 elapsed_s = (state.accumulated_ms or 0) // 1000
101 elapsed_str = (
102 f"{elapsed_s}s" if elapsed_s < 60 else f"{elapsed_s // 60}m {elapsed_s % 60}s"
103 )
104 name_str = colorize(state.loop_name, "1")
105 state_str = colorize(state.current_state, "34")
106 status_color = _STATUS_COLORS.get(state.status, "2")
107 display_status = "paused" if state.status == "interrupted" else state.status
108 status_str = colorize(f"[{display_status}]", status_color)
109 elapsed_colored = colorize(elapsed_str, "2")
110 print(
111 f" {name_str}: {state_str} (iteration {state.iteration})"
112 f" {status_str} {elapsed_colored}"
113 )
114 else:
115 # Multiple instances: show a grouped summary
116 name_str = colorize(loop_name_key, "1")
117 statuses = ", ".join(
118 colorize(
119 f"[{'paused' if s.status == 'interrupted' else s.status}]",
120 _STATUS_COLORS.get(s.status, "2"),
121 )
122 for s in group_states
123 )
124 count_str = colorize(f"({len(group_states)} instances)", "2")
125 print(f" {name_str}: {count_str} {statuses}")
126 return 0
128 builtin_only = getattr(args, "builtin", False)
130 def _rel_key(path: Path, base: Path) -> str:
131 """Relative-path identifier matching what `ll-loop run` accepts."""
132 return str(path.relative_to(base).with_suffix(""))
134 # Collect project loops (skipped when --builtin is set). Recurse into
135 # subdirectories (e.g. oracles/) and filter to runnable FSM definitions so
136 # library fragments under loops/lib/ stay hidden.
137 project_names: set[str] = set()
138 yaml_files: list[Path] = []
139 if not builtin_only and loops_dir.exists():
140 yaml_files = sorted(p for p in loops_dir.rglob("*.yaml") if is_runnable_loop(p))
141 project_names = {_rel_key(p, loops_dir) for p in yaml_files}
143 # Collect built-in loops (excluding those overridden by a project loop at
144 # the same relative path).
145 builtin_dir = get_builtin_loops_dir()
146 builtin_files: list[Path] = []
147 if builtin_dir.exists():
148 builtin_files = [
149 f
150 for f in sorted(builtin_dir.rglob("*.yaml"))
151 if is_runnable_loop(f) and _rel_key(f, builtin_dir) not in project_names
152 ]
154 if not yaml_files and not builtin_files:
155 if getattr(args, "json", False):
156 print_json([])
157 return 0
158 print("No loops available")
159 return 0
161 # Build combined metadata list
162 all_loops: list[dict[str, Any]] = []
163 for path in yaml_files:
164 meta = _load_loop_meta(path)
165 all_loops.append(
166 {"name": _rel_key(path, loops_dir), "path": path, "builtin": False, **meta}
167 )
168 for path in builtin_files:
169 meta = _load_loop_meta(path)
170 all_loops.append(
171 {"name": _rel_key(path, builtin_dir), "path": path, "builtin": True, **meta}
172 )
174 # Apply --category filter
175 category_filter = getattr(args, "category", None)
176 if category_filter:
177 all_loops = [lp for lp in all_loops if lp["category"] == category_filter]
179 # Apply --label filter (action="append" → list or None)
180 label_filters: list[str] = getattr(args, "label", None) or []
181 if label_filters:
182 all_loops = [
183 lp
184 for lp in all_loops
185 if any(lf.lower() in [lb.lower() for lb in lp["labels"]] for lf in label_filters)
186 ]
188 # Apply visibility filter. Default view shows only "public" loops, hiding
189 # delegated-only sub-loops ("internal") and demos/templates ("example").
190 # --all shows everything; --internal / --examples narrow to those tiers.
191 show_all = getattr(args, "all", False)
192 show_internal = getattr(args, "internal", False)
193 show_examples = getattr(args, "examples", False)
194 hidden_counts: dict[str, int] = {"internal": 0, "example": 0}
195 if not show_all:
196 if show_internal or show_examples:
197 wanted = set()
198 if show_internal:
199 wanted.add("internal")
200 if show_examples:
201 wanted.add("example")
202 all_loops = [lp for lp in all_loops if lp.get("visibility", "public") in wanted]
203 else:
204 for lp in all_loops:
205 vis = lp.get("visibility", "public")
206 if vis in hidden_counts:
207 hidden_counts[vis] += 1
208 all_loops = [lp for lp in all_loops if lp.get("visibility", "public") == "public"]
210 if not all_loops:
211 if getattr(args, "json", False):
212 print_json([])
213 return 0
214 print("No loops match the given filters")
215 return 0
217 if getattr(args, "json", False):
218 items: list[dict[str, Any]] = []
219 for lp in all_loops:
220 item: dict[str, Any] = {
221 "name": lp["name"],
222 "path": str(lp["path"]),
223 "category": lp["category"],
224 "labels": lp["labels"],
225 "visibility": lp.get("visibility", "public"),
226 "description": lp["description"],
227 }
228 if lp["builtin"]:
229 item["built_in"] = True
230 items.append(item)
231 print_json(items)
232 return 0
234 # Human-readable: group by category
235 buckets: dict[str, list[dict[str, Any]]] = {}
236 for lp in all_loops:
237 cat = lp["category"] or "uncategorized"
238 if cat not in buckets:
239 buckets[cat] = []
240 buckets[cat].append(lp)
242 # Sort categories; "uncategorized" always last
243 sorted_cats = sorted(c for c in buckets if c != "uncategorized")
244 if "uncategorized" in buckets:
245 sorted_cats.append("uncategorized")
247 # Compute max name width for column alignment
248 max_name_len = max((len(lp["name"]) for lp in all_loops), default=0)
249 name_col = max_name_len + 2 # padding after longest name
250 tw = terminal_width()
252 cats_printed = False
253 for cat in sorted_cats:
254 group = buckets[cat]
255 if cats_printed:
256 print() # blank line between category groups
257 cats_printed = True
258 print(colorize(f"{cat} ({len(group)}):", "1"))
259 for lp in group:
260 # Name: project loops get bold cyan, built-in loops get dimmer cyan
261 name_color = "36" if lp["builtin"] else "36;1"
262 name_str = colorize(lp["name"].ljust(name_col), name_color)
264 # Suffix: labels + [built-in] tag
265 suffix_parts: list[str] = []
266 if lp["labels"]:
267 for label in lp["labels"]:
268 suffix_parts.append(colorize(f"[{label}]", "2"))
269 if lp["builtin"]:
270 suffix_parts.append(colorize("[built-in]", "2"))
272 if suffix_parts:
273 suffix_raw = " " + " ".join(suffix_parts)
274 suffix_visible = len(strip_ansi(suffix_raw))
275 else:
276 suffix_raw = ""
277 suffix_visible = 0
279 # Available width for description: indent + name_col + " " + desc + suffix
280 avail = tw - 2 - name_col - 2 - suffix_visible
281 desc_text = lp["description"] or ""
282 if desc_text and avail < len(desc_text):
283 desc_text = _truncate(desc_text, max(avail, 20))
284 desc_str = f" {colorize(desc_text, '2')}" if desc_text else ""
286 print(f" {name_str}{desc_str}{suffix_raw}")
288 # Footer: surface hidden tiers and point users at the natural-language router
289 # rather than asking them to scan dozens of loops.
290 hidden_bits: list[str] = []
291 if hidden_counts.get("internal"):
292 hidden_bits.append(f"{hidden_counts['internal']} internal (--internal)")
293 if hidden_counts.get("example"):
294 hidden_bits.append(f"{hidden_counts['example']} example (--examples)")
295 if hidden_bits:
296 print()
297 print(colorize(f"Hidden: {', '.join(hidden_bits)} · all with --all", "2"))
298 print(
299 colorize(
300 'Not sure which loop? `ll-loop run loop-router --input goal="<what you want>"`',
301 "2",
302 )
303 )
304 return 0
307_EVENT_TYPE_WIDTH = 16 # width of "handoff_detected"
310def _truncate(text: str, max_len: int) -> str:
311 """Truncate text to max_len with ellipsis."""
312 if max_len < 1:
313 return ""
314 if len(text) <= max_len:
315 return text
316 return text[: max_len - 1] + "\u2026"
319def _format_history_event(
320 event: dict[str, Any], verbose: bool, width: int, full: bool = False
321) -> str | None:
322 """Format a single history event. Returns None to skip the event."""
323 raw_ts = event.get("ts", "")
324 try:
325 ts = datetime.fromisoformat(raw_ts).strftime("%H:%M:%S")
326 except (ValueError, TypeError):
327 ts = raw_ts[:8] if len(raw_ts) >= 8 else raw_ts.ljust(8)
329 event_type = event.get("event", "unknown")
331 if event_type == "action_output" and not verbose:
332 return None
334 ts_str = colorize(ts, "2")
335 etype_padded = event_type.ljust(_EVENT_TYPE_WIDTH)
336 etype_color = "0"
337 detail = ""
338 extra_lines: list[str] = []
340 # Indentation prefix for verbose sub-lines (aligns under event detail column)
341 _indent = " " * (8 + 2 + _EVENT_TYPE_WIDTH + 2)
343 if event_type == "loop_start":
344 etype_color = "1"
345 detail = event.get("loop", "")
347 elif event_type == "loop_complete":
348 etype_color = "1"
349 final_state = event.get("final_state", "")
350 iterations = event.get("iterations", "")
351 terminated_by = event.get("terminated_by", "")
352 detail = f"{final_state} {iterations} iter [{terminated_by}]"
354 elif event_type == "loop_resume":
355 etype_color = "1"
356 from_state = event.get("from_state", "")
357 iteration = event.get("iteration", "")
358 detail = f"from={from_state} iter={iteration}"
360 elif event_type == "state_enter":
361 etype_color = "34"
362 state = event.get("state", "")
363 iteration = event.get("iteration", "")
364 detail = f"{colorize(state, '1')} (iter {iteration})"
366 elif event_type == "action_start":
367 action = event.get("action", "")
368 is_prompt = event.get("is_prompt", False)
369 kind_label = "prompt" if is_prompt else "shell"
370 kind_str = colorize(f"[{kind_label}]", "2")
371 first_line = (
372 next((ln.strip() for ln in action.splitlines() if ln.strip()), "")
373 if is_prompt
374 else action
375 )
376 avail = width - 8 - 2 - _EVENT_TYPE_WIDTH - 2 - len(kind_label) - 2 - 2
377 detail = f"{_truncate(first_line, max(avail, 20))} {kind_str}"
379 elif event_type == "action_output":
380 # Only reached in verbose mode
381 etype_color = "2"
382 detail = colorize("\u2502 " + event.get("line", ""), "2")
384 elif event_type == "action_complete":
385 exit_code = event.get("exit_code", 0)
386 duration_ms = event.get("duration_ms", 0)
387 if exit_code == 0:
388 etype_color = "2"
389 status_str = colorize("\u2713", "32")
390 else:
391 etype_color = "38;5;208"
392 status_str = colorize(f"\u2717 exit={exit_code}", "38;5;208")
393 detail = f"{status_str} {duration_ms}ms"
394 is_prompt = event.get("is_prompt", False)
395 session_jsonl = event.get("session_jsonl") if is_prompt else None
396 if session_jsonl:
397 session_display = session_jsonl if verbose else os.path.basename(session_jsonl)
398 detail += f" session={colorize(session_display, '2')}"
399 if verbose:
400 output_preview = event.get("output_preview", "")
401 if output_preview:
402 avail_w = width - len(_indent) - 2
403 preview_text = (
404 output_preview if full else _truncate(output_preview, max(avail_w, 40))
405 )
406 for preview_line in preview_text.splitlines()[:5]:
407 extra_lines.append(colorize(_indent + "\u2502 " + preview_line, "2"))
409 elif event_type == "evaluate":
410 verdict = event.get("verdict", "")
411 confidence = event.get("confidence", "")
412 reason = event.get("reason", "")
413 if verdict == "yes":
414 etype_color = "32"
415 verdict_str = colorize("\u2713 yes", "32")
416 else:
417 etype_color = "38;5;208"
418 verdict_str = colorize(f"\u2717 {verdict}", "38;5;208")
419 conf_part = f" confidence={confidence}" if confidence != "" else ""
420 avail = width - 8 - 2 - _EVENT_TYPE_WIDTH - 2 - len("\u2713 yes") - len(conf_part) - 2
421 reason_part = f" {_truncate(reason, max(avail, 20))}" if reason else ""
422 detail = f"{verdict_str}{conf_part}{reason_part}"
423 if verbose:
424 llm_model = event.get("llm_model", "")
425 llm_latency_ms = event.get("llm_latency_ms", "")
426 llm_prompt = event.get("llm_prompt", "")
427 llm_raw_output = event.get("llm_raw_output", "")
428 if llm_model or llm_prompt:
429 meta_parts = []
430 if llm_model:
431 meta_parts.append(f"model={llm_model}")
432 if llm_latency_ms != "":
433 meta_parts.append(f"latency={llm_latency_ms}ms")
434 meta_str = " ".join(meta_parts)
435 extra_lines.append(
436 colorize(_indent + colorize("LLM Call", "2") + " " + meta_str, "2")
437 )
438 avail_w = width - len(_indent) - len("Prompt: ") - 2
439 if llm_prompt:
440 prompt_text = llm_prompt if full else _truncate(llm_prompt, max(avail_w, 40))
441 extra_lines.append(colorize(_indent + "Prompt: " + prompt_text, "2"))
442 if llm_raw_output:
443 resp_text = (
444 llm_raw_output if full else _truncate(llm_raw_output, max(avail_w, 40))
445 )
446 extra_lines.append(colorize(_indent + "Response: " + resp_text, "2"))
448 elif event_type == "route":
449 etype_color = "2"
450 from_state = event.get("from", "")
451 to_state = event.get("to", "")
452 detail = f"{from_state} \u2192 {colorize(to_state, '34')}"
454 elif event_type == "handoff_detected":
455 etype_color = "33"
456 detail = f"state={event.get('state', '')} iter={event.get('iteration', '')}"
458 else:
459 details = {k: v for k, v in event.items() if k not in ("event", "ts")}
460 detail = " ".join(f"{k}={v}" for k, v in details.items())
462 etype_str = colorize(etype_padded, etype_color)
463 main_line = f"{ts_str} {etype_str} {detail}"
464 if extra_lines:
465 return "\n".join([main_line] + extra_lines)
466 return main_line
469def _format_duration(ms: int) -> str:
470 """Format milliseconds as a human-readable duration."""
471 if ms < 1000:
472 return f"{ms}ms"
473 s = ms // 1000
474 if s < 60:
475 return f"{s}s"
476 m, s = divmod(s, 60)
477 if m < 60:
478 return f"{m}m{s:02d}s"
479 h, m = divmod(m, 60)
480 return f"{h}h{m:02d}m{s:02d}s"
483def _list_archived_runs(loop_name: str, loops_dir: Path, as_json: bool) -> int:
484 """List archived runs for a loop."""
485 import json as _json
487 from little_loops.fsm.persistence import HISTORY_DIR, LoopState
489 history_base = loops_dir / HISTORY_DIR
490 if not history_base.exists():
491 print(f"No history for: {loop_name}")
492 return 0
494 # Flat layout: run dirs are <run_id>-<loop_name> directly under .history/
495 suffix = f"-{loop_name}"
496 runs: list[tuple[str, LoopState | None]] = []
497 for run_dir in sorted(history_base.iterdir(), key=lambda d: d.name, reverse=True):
498 if not run_dir.is_dir() or not run_dir.name.endswith(suffix):
499 continue
500 run_id = run_dir.name[: -len(suffix)]
501 state_file = run_dir / "state.json"
502 state: LoopState | None = None
503 if state_file.exists():
504 try:
505 data = _json.loads(state_file.read_text())
506 state = LoopState.from_dict(data)
507 except (ValueError, KeyError):
508 pass
509 runs.append((run_id, state))
511 if not runs:
512 print(f"No history for: {loop_name}")
513 return 0
515 if as_json:
516 print(
517 _json.dumps(
518 [
519 {
520 "run_id": rid,
521 "status": s.status if s else None,
522 "started_at": s.started_at if s else None,
523 "iterations": s.iteration if s else None,
524 "duration_ms": s.accumulated_ms if s else None,
525 }
526 for rid, s in runs
527 ],
528 indent=2,
529 )
530 )
531 return 0
533 status_colors = {
534 "completed": "\033[32m",
535 "failed": "\033[31m",
536 "interrupted": "\033[33m",
537 "awaiting_continuation": "\033[36m",
538 "timed_out": "\033[33m",
539 "running": "\033[34m",
540 }
541 reset = "\033[0m"
543 print(f"Archived runs for: {loop_name} ({len(runs)} total)")
544 print()
546 for run_id, state in runs:
547 if state is not None:
548 color = status_colors.get(state.status, "")
549 status_str = f"{color}{state.status}{reset}"
550 duration_str = _format_duration(state.accumulated_ms) if state.accumulated_ms else "?"
551 started = state.started_at[:19].replace("T", " ") if state.started_at else "?"
552 iters = f"{state.iteration} iters"
553 else:
554 status_str = "unknown"
555 duration_str = "?"
556 started = "?"
557 iters = "?"
558 print(f" {run_id} {status_str} {started} {iters} {duration_str}")
560 print()
561 print(f"To view events: ll-loop history {loop_name} <run-id>")
562 return 0
565def cmd_history(
566 loop_name: str,
567 run_id: str | None,
568 args: argparse.Namespace,
569 loops_dir: Path,
570) -> int:
571 """Show loop history.
573 Without run_id: lists all archived runs with status and duration.
574 With run_id: shows events for that specific archived run.
575 """
576 tail = getattr(args, "tail", 50)
577 full = getattr(args, "full", False)
578 verbose = getattr(args, "verbose", False) or full
579 as_json = getattr(args, "json", False)
581 if run_id is None:
582 return _list_archived_runs(loop_name, loops_dir, as_json)
584 # Show events for a specific archived run
585 from little_loops.fsm.persistence import get_archived_events
587 events = get_archived_events(loop_name, run_id, loops_dir)
589 if not events:
590 print(f"No events found for run {run_id} of loop {loop_name}")
591 return 1
593 w = terminal_width()
594 if not verbose:
595 events = [e for e in events if e.get("event") != "action_output"]
597 # Apply optional filters (before --tail slice)
598 event_filter = getattr(args, "event", None)
599 if event_filter:
600 events = [e for e in events if e.get("event") == event_filter]
602 state_filter = getattr(args, "state", None)
603 if state_filter:
604 events = [
605 e
606 for e in events
607 if (
608 e.get("state") == state_filter
609 or e.get("from") == state_filter
610 or e.get("to") == state_filter
611 )
612 ]
614 since_str = getattr(args, "since", None)
615 if since_str:
616 from datetime import timedelta
618 from little_loops.text_utils import parse_duration
620 cutoff = datetime.now() - timedelta(seconds=parse_duration(since_str))
621 events = [
622 e
623 for e in events
624 if datetime.fromisoformat(e["ts"].replace("Z", "+00:00")).replace(tzinfo=None) >= cutoff
625 ]
627 if as_json:
628 print_json(events[-tail:])
629 return 0
630 for event in events[-tail:]:
631 line = _format_history_event(event, verbose, w, full=full)
632 if line is not None:
633 print(line)
635 return 0
638def cmd_audit_meta(loop_name: str, args: argparse.Namespace, loops_dir: Path) -> int:
639 """Summarize meta-eval.jsonl agreement stats from all archived runs of a loop.
641 Reads meta-eval.jsonl from each archived run, computes:
642 - Total iterations with llm_structured evaluate events
643 - Agreement rate (agreed / total)
644 - Mean diff size (files_changed) per verdict
645 - Divergence flags:
646 - agreed=false streak >=3 → "LLM optimistic drift detected"
647 - agreed=true with files_changed==0 streak >=3 → "Trivial agreement detected"
649 Returns 0 if no flags triggered, 1 if any threshold crossed.
650 """
651 import json as _json
653 from little_loops.fsm.persistence import HISTORY_DIR
655 history_root = loops_dir / HISTORY_DIR
656 if not history_root.exists():
657 print(f"No history for: {loop_name}")
658 return 0
660 suffix = f"-{loop_name}"
661 all_entries: list[dict[str, Any]] = []
663 for run_dir in sorted(history_root.iterdir(), key=lambda d: d.name):
664 if not run_dir.is_dir() or not run_dir.name.endswith(suffix):
665 continue
666 meta_eval_file = run_dir / "meta-eval.jsonl"
667 if not meta_eval_file.exists():
668 continue
669 for line in meta_eval_file.read_text(encoding="utf-8").splitlines():
670 line = line.strip()
671 if line:
672 try:
673 all_entries.append(_json.loads(line))
674 except _json.JSONDecodeError:
675 pass
677 if not all_entries:
678 print(f"No meta-eval data for: {loop_name}")
679 return 0
681 total = len(all_entries)
682 agreed_count = sum(1 for e in all_entries if e.get("agreed") is True)
683 agreement_rate = agreed_count / total if total else 0.0
685 # Mean diff size per verdict
686 agreed_sizes = [
687 e.get("diff_stats", {}).get("files_changed", 0) or 0
688 for e in all_entries
689 if e.get("agreed") is True
690 ]
691 disagreed_sizes = [
692 e.get("diff_stats", {}).get("files_changed", 0) or 0
693 for e in all_entries
694 if e.get("agreed") is False
695 ]
696 mean_agreed = sum(agreed_sizes) / len(agreed_sizes) if agreed_sizes else 0.0
697 mean_disagreed = sum(disagreed_sizes) / len(disagreed_sizes) if disagreed_sizes else 0.0
699 # Detect streaks
700 flags: list[str] = []
701 optimistic_streak = 0
702 trivial_streak = 0
703 max_optimistic = 0
704 max_trivial = 0
706 for entry in all_entries:
707 agreed = entry.get("agreed")
708 files_changed = (entry.get("diff_stats") or {}).get("files_changed", 0) or 0
710 if agreed is False:
711 optimistic_streak += 1
712 trivial_streak = 0
713 elif agreed is True and files_changed == 0:
714 trivial_streak += 1
715 optimistic_streak = 0
716 else:
717 optimistic_streak = 0
718 trivial_streak = 0
720 max_optimistic = max(max_optimistic, optimistic_streak)
721 max_trivial = max(max_trivial, trivial_streak)
723 if max_optimistic >= 3:
724 flags.append(f"LLM optimistic drift detected (streak={max_optimistic})")
725 if max_trivial >= 3:
726 flags.append(f"Trivial agreement detected (streak={max_trivial})")
728 as_json = getattr(args, "json", False)
729 if as_json:
730 result = {
731 "loop": loop_name,
732 "total_entries": total,
733 "agreed_count": agreed_count,
734 "agreement_rate": round(agreement_rate, 4),
735 "mean_files_changed_when_agreed": round(mean_agreed, 2),
736 "mean_files_changed_when_disagreed": round(mean_disagreed, 2),
737 "max_optimistic_streak": max_optimistic,
738 "max_trivial_streak": max_trivial,
739 "flags": flags,
740 }
741 print_json(result)
742 else:
743 print(f"Meta-eval audit: {loop_name}")
744 print(f" Total entries: {total}")
745 print(f" Agreed: {agreed_count}/{total} ({agreement_rate:.0%})")
746 print(f" Mean Δfiles agreed: {mean_agreed:.1f}")
747 print(f" Mean Δfiles disagreed: {mean_disagreed:.1f}")
748 if flags:
749 print()
750 for flag in flags:
751 print(f" ⚠ {flag}")
752 else:
753 print(" No divergence flags.")
755 if as_json:
756 return 0 # let caller inspect JSON for flags; non-zero kills provider.runCli
757 return 1 if flags else 0
760def cmd_diagnose_evaluators(loop_name: str, args: argparse.Namespace, loops_dir: Path) -> int:
761 """Detect non-discriminating evaluators from run history.
763 Walks .loops/.history/*-{loop_name}/events.jsonl, computes per-state
764 Bernoulli variance p*(1-p) on verdicts, flags states below threshold
765 with pattern-matched recommendations.
767 Returns 0 if no states flagged, 1 if any low-variance evaluators found.
768 """
769 from little_loops.analytics.variance import compute_evaluator_variance
771 threshold = getattr(args, "threshold", 0.05)
772 min_runs = getattr(args, "min_runs", 10)
773 as_json = getattr(args, "json", False)
775 report = compute_evaluator_variance(loop_name, loops_dir, threshold, min_runs)
777 if report is None:
778 if not (loops_dir / ".history").exists():
779 print(f"No history for: {loop_name}")
780 elif len(list((loops_dir / ".history").glob(f"*-{loop_name}"))) < min_runs:
781 runs_found = len(list((loops_dir / ".history").glob(f"*-{loop_name}")))
782 print(
783 f"Insufficient history for: {loop_name} "
784 f"(found {runs_found} run(s), need {min_runs})"
785 )
786 else:
787 print(f"No evaluate events for: {loop_name}")
788 return 0
790 flagged = [s for s in report.states if s.variance < threshold]
792 if as_json:
793 print_json(report.to_dict())
794 else:
795 print(f"Evaluator Variance Report (n={report.total_runs} runs)")
796 if not report.states:
797 print(" No evaluator states found in run history.")
798 for state in report.states:
799 disc_label = "✓ discriminating" if state.variance >= threshold else "⚠ low variance"
800 ci_str = ""
801 if state.ci is not None:
802 ci_str = f" CI=[{state.ci[0]:.2f}, {state.ci[1]:.2f}]"
803 print(
804 f" {state.state:20s} pass_rate={state.pass_rate:.2f} "
805 f"variance={state.variance:.2f}{ci_str} {disc_label}"
806 )
807 if state.recommendation:
808 for line in state.recommendation.split("\n"):
809 print(f" {line.strip()}")
811 if as_json:
812 return 0
813 return 1 if flagged else 0
816def cmd_calibrate_budget(loop_name: str, args: argparse.Namespace, loops_dir: Path) -> int:
817 """Report per-evaluator Bernoulli variance to guide max_iterations calibration.
819 Calls compute_evaluator_variance() from analytics.variance and reports
820 p*(1-p) per evaluator state. Variance below threshold signals that
821 increasing max_iterations is unlikely to help; fix the evaluator first.
823 Returns 0 if all evaluators are healthy, 1 if any are flagged.
824 """
825 from little_loops.analytics.variance import compute_evaluator_variance
827 threshold = getattr(args, "threshold", 0.05)
828 min_runs = getattr(args, "min_runs", 10)
829 as_json = getattr(args, "json", False)
831 report = compute_evaluator_variance(loop_name, loops_dir, threshold, min_runs)
833 if report is None:
834 if not (loops_dir / ".history").exists():
835 print(f"No history for: {loop_name}")
836 elif len(list((loops_dir / ".history").glob(f"*-{loop_name}"))) < min_runs:
837 runs_found = len(list((loops_dir / ".history").glob(f"*-{loop_name}")))
838 print(
839 f"Insufficient history for: {loop_name} "
840 f"(found {runs_found} run(s), need {min_runs})"
841 )
842 else:
843 print(f"No evaluate events for: {loop_name}")
844 return 0
846 flagged = [s for s in report.states if s.variance < threshold]
848 if as_json:
849 print_json(report.to_dict())
850 return 0
852 print(f"Loop: {loop_name}")
853 for state in report.states:
854 evaluator_type = state.evaluator_type or "unknown"
855 print(f"Evaluator: {state.state} ({evaluator_type})")
856 ci_str = ""
857 if state.ci is not None:
858 ci_str = f" CI=[{state.ci[0]:.2f}, {state.ci[1]:.2f}]"
859 if state.variance < threshold:
860 print(
861 f" Variance p*(1-p): {state.variance:.2f}{ci_str}"
862 f" ⚠ WARN: below {threshold} threshold"
863 f" — fix evaluator before increasing max_iterations"
864 )
865 else:
866 print(f" Variance p*(1-p): {state.variance:.2f}{ci_str} ✓ OK")
868 return 1 if flagged else 0
871def cmd_promote_baseline(loop_name: str, args: argparse.Namespace, loops_dir: Path) -> int:
872 """Promote the latest run's action output as the new comparator baseline.
874 Reads action_output events from the most recent .history entry and writes
875 the concatenated output to .loops/baselines/<loop>/output.txt.
876 """
877 import json as _json
879 history_base = loops_dir / ".history"
880 if not history_base.exists():
881 print(f"No history for: {loop_name}")
882 return 1
884 suffix = f"-{loop_name}"
885 matched = sorted(
886 [d for d in history_base.iterdir() if d.is_dir() and d.name.endswith(suffix)],
887 key=lambda d: d.name,
888 reverse=True,
889 )
890 if not matched:
891 print(f"No history for: {loop_name}")
892 return 1
894 latest = matched[0]
895 events_file = latest / "events.jsonl"
896 if not events_file.exists():
897 print(f"No events.jsonl in latest run: {latest.name}")
898 return 1
900 lines = []
901 with open(events_file) as f:
902 for raw in f:
903 raw = raw.strip()
904 if not raw:
905 continue
906 try:
907 event = _json.loads(raw)
908 except _json.JSONDecodeError:
909 continue
910 if event.get("type") == "action_output":
911 line = event.get("line", "")
912 if line:
913 lines.append(line)
915 if not lines:
916 print(f"No action_output events in latest run: {latest.name}")
917 return 1
919 target = Path(loops_dir) / "baselines" / loop_name / "output.txt"
920 target.parent.mkdir(parents=True, exist_ok=True)
921 target.write_text("\n".join(lines))
922 print(f"Promoted baseline for {loop_name}: {target}")
923 return 0
926# ---------------------------------------------------------------------------
927# FSM diagram renderer — delegated to layout module (re-exported above)
928# ---------------------------------------------------------------------------
931# ---------------------------------------------------------------------------
932# State overview table
933# ---------------------------------------------------------------------------
936def _compact_transitions(state: StateConfig) -> str:
937 """Return a compact transition string for the overview table."""
938 raw: list[tuple[str, str]] = []
939 for label, target in [
940 ("yes", state.on_yes),
941 ("no", state.on_no),
942 ("error", state.on_error),
943 ("partial", state.on_partial),
944 ("next", state.next),
945 ]:
946 if target:
947 raw.append((label, target))
948 if state.route:
949 for verdict, target in state.route.routes.items():
950 raw.append((verdict, target))
951 if state.route.default:
952 raw.append(("_", state.route.default))
953 if not raw:
954 return "\u2014"
955 # Group by target, preserving first-seen order
956 seen: list[str] = []
957 by_target: dict[str, list[str]] = {}
958 for label, target in raw:
959 if target not in by_target:
960 seen.append(target)
961 by_target[target] = []
962 by_target[target].append(label)
963 return ", ".join(f"{'/'.join(by_target[t])}\u2192{t}" for t in seen)
966def _print_state_overview_table(fsm: FSMLoop) -> None:
967 """Print a compact summary table of all states."""
968 rows: list[tuple[str, str, str, str]] = []
969 for name, state in fsm.states.items():
970 # State name column
971 state_col = f"\u2192 {name}" if name == fsm.initial else f" {name}"
973 # Type column
974 if state.terminal:
975 type_col = "\u2014"
976 elif state.action_type:
977 type_col = state.action_type
978 elif state.action:
979 type_col = "shell"
980 else:
981 type_col = "\u2014"
983 # Action preview column
984 if state.terminal:
985 action_col = "(terminal)"
986 elif state.action:
987 src_lines = [ln.rstrip() for ln in state.action.strip().splitlines() if ln.rstrip()]
988 action_col = src_lines[0] if src_lines else "\u2014"
989 else:
990 action_col = "\u2014"
992 # Transitions column
993 trans_col = _compact_transitions(state)
994 rows.append((state_col, type_col, action_col, trans_col))
996 if not rows:
997 return
999 tw = terminal_width()
1000 headers = ("State", "Type", "Action Preview", "Transitions")
1001 col0_w = max(len(headers[0]), max(len(r[0]) for r in rows))
1002 col1_w = max(len(headers[1]), max(len(r[1]) for r in rows))
1003 # Remaining width split between action preview and transitions
1004 fixed = col0_w + col1_w + 10 # margins + separators
1005 remaining = max(20, tw - fixed)
1006 col2_w = min(50, max(len(headers[2]), max(len(r[2]) for r in rows)), remaining * 3 // 5)
1007 col2_w = max(10, col2_w)
1008 col3_w = max(10, remaining - col2_w)
1010 print(f" {headers[0]:<{col0_w}} {headers[1]:<{col1_w}} {headers[2]:<{col2_w}} {headers[3]}")
1011 dash = "\u2500"
1012 print(f" {dash * col0_w} {dash * col1_w} {dash * col2_w} {dash * col3_w}")
1013 for state_col, type_col, action_col, trans_col in rows:
1014 if len(action_col) > col2_w:
1015 action_col = action_col[: col2_w - 1] + "\u2026"
1016 if len(trans_col) > col3_w:
1017 trans_col = trans_col[: col3_w - 1] + "\u2026"
1018 colored_type = colorize(type_col, "2") if type_col == "\u2014" else type_col
1019 print(
1020 f" {state_col:<{col0_w}} {colored_type:<{col1_w}} "
1021 f"{action_col:<{col2_w}} {trans_col}"
1022 )
1025# ---------------------------------------------------------------------------
1026# cmd_show
1027# ---------------------------------------------------------------------------
1029_EVALUATE_TYPE_DISPLAY: dict[str, str] = {
1030 "llm": "LLM",
1031 "llm_structured": "LLM (structured)",
1032 "exit_code": "exit code",
1033 "output_numeric": "numeric",
1034 "output_contains": "contains",
1035 "output_json": "JSON",
1036 "convergence": "convergence",
1037 "diff_stall": "diff stall",
1038 "action_stall": "action stall",
1039 "comparator": "blind comparator",
1040}
1043def _humanize_evaluate_type(ev_type: str) -> str:
1044 return _EVALUATE_TYPE_DISPLAY.get(ev_type, ev_type)
1047def cmd_show(
1048 loop_name: str,
1049 args: argparse.Namespace,
1050 loops_dir: Path,
1051 logger: Logger,
1052) -> int:
1053 """Show loop details and structure."""
1054 try:
1055 path = resolve_loop_path(loop_name, loops_dir)
1056 fsm, spec = load_loop_with_spec(loop_name, loops_dir, logger)
1057 except FileNotFoundError as e:
1058 logger.error(str(e))
1059 return 1
1060 except ValueError as e:
1061 logger.error(f"Invalid loop: {e}")
1062 return 1
1064 if getattr(args, "json", False) and getattr(args, "show_diagrams", None) is not None:
1065 logger.error("--json and --show-diagrams are mutually exclusive")
1066 return 1
1068 if getattr(args, "json", False):
1069 data = fsm.to_dict()
1070 if getattr(args, "resolved", False):
1071 for state in data.get("states", {}).values():
1072 if "loop" in state:
1073 try:
1074 child_path = resolve_loop_path(state["loop"], loops_dir)
1075 child_fsm, _ = load_and_validate(child_path)
1076 state["_subloop"] = child_fsm.to_dict().get("states", {})
1077 except (FileNotFoundError, ValueError):
1078 pass
1079 print_json(data)
1080 return 0
1082 tw = terminal_width()
1084 # Compute stats for header
1085 n_states = len(fsm.states)
1086 n_transitions = sum(
1087 bool(s.on_yes)
1088 + bool(s.on_no)
1089 + bool(s.on_error)
1090 + bool(s.on_partial)
1091 + bool(s.next)
1092 + bool(s.on_maintain)
1093 + (len(s.route.routes) + bool(s.route.default) if s.route else 0)
1094 for s in fsm.states.values()
1095 )
1097 # --- Compact metadata header ---
1098 # Line 1: ── name ───────── N states · M transitions ──
1099 stats_parts: list[str] = []
1100 stats_parts.append(f"{n_states} states")
1101 stats_parts.append(f"{n_transitions} transitions")
1102 stats_str = " \u00b7 ".join(stats_parts)
1104 header_left = f"\u2500\u2500 {loop_name} "
1105 header_right = f" {stats_str} \u2500\u2500"
1106 dashes = "\u2500" * max(0, tw - len(header_left) - len(header_right))
1107 print(f"{header_left}{dashes}{header_right}")
1109 # Line 2: source · max: N iter · handoff: X [· optional fields]
1110 config_parts: list[str] = [str(path), f"max: {fsm.max_iterations} iter"]
1111 config_parts.append(f"handoff: {fsm.on_handoff}")
1112 if fsm.on_max_iterations is not None:
1113 config_parts.append(f"on_max_iterations: {fsm.on_max_iterations}")
1114 if fsm.timeout:
1115 config_parts.append(f"timeout: {fsm.timeout}s")
1116 if fsm.backoff:
1117 config_parts.append(f"backoff: {fsm.backoff}s")
1118 if fsm.maintain:
1119 config_parts.append("maintain: yes")
1120 if fsm.context:
1121 config_parts.append(f"context: {', '.join(fsm.context.keys())}")
1122 if fsm.scope:
1123 config_parts.append(f"scope: {', '.join(fsm.scope)}")
1124 llm = fsm.llm
1125 llm_parts = []
1126 if llm.model != "sonnet":
1127 llm_parts.append(f"model={llm.model}")
1128 if llm.max_tokens != 256:
1129 llm_parts.append(f"max_tokens={llm.max_tokens}")
1130 if llm.timeout != 30:
1131 llm_parts.append(f"timeout={llm.timeout}s")
1132 if llm_parts:
1133 config_parts.append(f"llm: {', '.join(llm_parts)}")
1134 if fsm.config is not None:
1135 cfg_parts = []
1136 if fsm.config.handoff_threshold is not None:
1137 cfg_parts.append(f"handoff_threshold={fsm.config.handoff_threshold}")
1138 if fsm.config.readiness_threshold is not None:
1139 cfg_parts.append(f"readiness_threshold={fsm.config.readiness_threshold}")
1140 if fsm.config.outcome_threshold is not None:
1141 cfg_parts.append(f"outcome_threshold={fsm.config.outcome_threshold}")
1142 if fsm.config.max_continuations is not None:
1143 cfg_parts.append(f"max_continuations={fsm.config.max_continuations}")
1144 if cfg_parts:
1145 config_parts.append(f"config: {', '.join(cfg_parts)}")
1146 imports = spec.get("import", [])
1147 if imports:
1148 config_parts.append(f"imports: {', '.join(imports)}")
1149 print(" " + " \u00b7 ".join(config_parts))
1151 # --- Description ---
1152 description = spec.get("description", "").strip()
1153 if description:
1154 print()
1155 print("Description:")
1156 for line in description.splitlines():
1157 print(f" {line}")
1159 # --- ASCII FSM Diagram ---
1160 verbose = getattr(args, "verbose", False)
1161 from pathlib import Path
1163 from little_loops.config import BRConfig
1165 badges = BRConfig(Path.cwd()).loops.glyphs.to_dict()
1166 facets = resolve_facets(args)
1167 print()
1168 print("Diagram:")
1169 if facets is None:
1170 diagram = _render_fsm_diagram(fsm, verbose=verbose, badges=badges)
1171 else:
1172 diagram = _render_fsm_diagram(
1173 fsm,
1174 badges=badges,
1175 mode=facets.scope,
1176 suppress_labels=not facets.edge_labels,
1177 title_only=facets.state_detail == "title",
1178 )
1179 if diagram:
1180 print(diagram)
1182 # --- State overview table ---
1183 print()
1184 _print_state_overview_table(fsm)
1186 # --- States & Transitions (verbose only) ---
1187 if verbose:
1188 print()
1189 print("States:")
1190 first_state = True
1191 for name, state in fsm.states.items():
1192 if not first_state:
1193 print()
1194 first_state = False
1196 # Improved state section header: ── name ──── MARKERS · type ──
1197 right_parts = []
1198 if name == fsm.initial:
1199 right_parts.append("INITIAL")
1200 if state.terminal:
1201 right_parts.append("TERMINAL")
1202 if state.action_type:
1203 right_parts.append(state.action_type)
1204 right_info = " \u00b7 ".join(right_parts)
1205 inner_left = f"\u2500\u2500 {name} "
1206 inner_right = f" {right_info} \u2500\u2500" if right_info else " \u2500\u2500"
1207 fill = "\u2500" * max(0, tw - 2 - len(inner_left) - len(inner_right))
1208 print(f" {inner_left}{fill}{inner_right}")
1210 if state.action:
1211 if verbose:
1212 indented = "\n ".join(state.action.strip().splitlines())
1213 print(f" action:\n {indented}")
1214 elif state.action_type == "prompt":
1215 lines_act = state.action.strip().splitlines()
1216 preview = "\n ".join(lines_act[:3])
1217 if len(lines_act) > 3 or len(state.action) > 200:
1218 preview += " ..."
1219 print(f" action:\n {preview}")
1220 else: # shell, slash_command, or None
1221 action_display = (
1222 state.action[:70] + "..." if len(state.action) > 70 else state.action
1223 )
1224 print(f" action: {action_display}")
1225 if state.evaluate:
1226 ev = state.evaluate
1227 print(f" evaluate: {_humanize_evaluate_type(ev.type)}")
1228 if ev.prompt:
1229 if verbose:
1230 print(" prompt:")
1231 for line in ev.prompt.strip().splitlines():
1232 print(f" \u2502 {line}")
1233 else:
1234 ev_lines = ev.prompt.strip().splitlines()
1235 preview = ev_lines[0][:100] + (
1236 " ..." if len(ev_lines) > 1 or len(ev_lines[0]) > 100 else ""
1237 )
1238 print(f" prompt: {preview}")
1239 if ev.min_confidence != 0.5:
1240 print(f" min_confidence: {ev.min_confidence}")
1241 if ev.operator:
1242 print(f" operator: {ev.operator} {ev.target}")
1243 if ev.pattern:
1244 print(f" pattern: {ev.pattern}")
1245 if state.capture:
1246 print(f" capture: {state.capture}")
1247 if state.timeout:
1248 print(f" timeout: {state.timeout}s")
1249 # Collect (label, target) pairs
1250 raw_transitions: list[tuple[str, str]] = []
1251 for label, target in [
1252 ("yes", state.on_yes),
1253 ("no", state.on_no),
1254 ("error", state.on_error),
1255 ("partial", state.on_partial),
1256 ("next", state.next),
1257 ("maintain", state.on_maintain),
1258 ]:
1259 if target:
1260 raw_transitions.append((label, target))
1261 if state.route:
1262 for verdict, target in state.route.routes.items():
1263 raw_transitions.append((verdict, target))
1264 if state.route.default:
1265 raw_transitions.append(("_", state.route.default))
1266 # Group by target, preserving first-seen order
1267 target_labels: dict[str, list[str]] = {}
1268 seen_targets: list[str] = []
1269 for label, target in raw_transitions:
1270 if target not in target_labels:
1271 target_labels[target] = []
1272 seen_targets.append(target)
1273 target_labels[target].append(label)
1274 transitions = [
1275 f"{_colorize_label('/'.join(target_labels[t]))} \u2500\u2500\u2192 {t}"
1276 for t in seen_targets
1277 ]
1278 if transitions:
1279 print(" Transitions:")
1280 for t in transitions:
1281 print(f" {t}")
1283 # --- Commands ---
1284 print()
1285 print("Commands:")
1286 if fsm.commands:
1287 cmds = [(e.cmd, e.comment) for e in fsm.commands]
1288 else:
1289 cmds = [
1290 (f"ll-loop run {loop_name}", "run"),
1291 (f"ll-loop test {loop_name}", "single test iteration"),
1292 (f"ll-loop stop {loop_name}", "stop a running loop"),
1293 (f"ll-loop status {loop_name}", "check if running"),
1294 (f"ll-loop history {loop_name}", "execution history"),
1295 ]
1296 col_width = max(len(c) for c, _ in cmds) + 2
1297 for cmd, comment in cmds:
1298 print(f" {cmd:<{col_width}} # {comment}")
1300 return 0
1303def cmd_fragments(
1304 lib_path: str,
1305 args: argparse.Namespace,
1306 loops_dir: Path,
1307 logger: Logger,
1308) -> int:
1309 """List fragments in a library file with their descriptions.
1311 Resolves the library path relative to loops_dir or the built-in loops directory,
1312 then prints a table of fragment names and their description fields.
1313 """
1314 import yaml
1316 # Resolve path: absolute/direct → loops_dir-relative → builtin
1317 p = Path(lib_path)
1318 if not p.exists():
1319 candidate = loops_dir / lib_path
1320 if candidate.exists():
1321 p = candidate
1322 else:
1323 builtin_candidate = get_builtin_loops_dir() / lib_path
1324 if builtin_candidate.exists():
1325 p = builtin_candidate
1326 else:
1327 logger.error(f"Fragment library not found: {lib_path}")
1328 return 1
1330 try:
1331 with open(p) as f:
1332 data = yaml.safe_load(f) or {}
1333 except Exception as e:
1334 logger.error(f"Failed to load library: {e}")
1335 return 1
1337 fragments: dict[str, Any] = data.get("fragments", {})
1338 if not fragments:
1339 print(f"No fragments defined in: {p}")
1340 return 0
1342 tw = terminal_width()
1343 print(colorize(f"Fragments in {p.name} ({len(fragments)}):", "1"))
1344 print()
1346 max_name_len = max(len(name) for name in fragments)
1347 name_col_w = max(max_name_len, 8)
1348 desc_col_w = max(20, tw - name_col_w - 6)
1350 header_name = "Fragment".ljust(name_col_w)
1351 print(f" {colorize(header_name, '1')} Description")
1352 print(f" {'─' * name_col_w} {'─' * desc_col_w}")
1354 for name, frag in fragments.items():
1355 raw_desc = frag.get("description", "") if isinstance(frag, dict) else ""
1356 raw_desc = raw_desc.strip()
1357 first_line = raw_desc.splitlines()[0] if raw_desc else ""
1358 if first_line and len(first_line) > desc_col_w:
1359 first_line = first_line[: desc_col_w - 1] + "\u2026"
1360 desc_str = first_line if first_line else colorize("(no description)", "2")
1361 padded_name = name.ljust(name_col_w)
1362 print(f" {colorize(padded_name, '36;1')} {desc_str}")
1364 return 0