Coverage for little_loops / cli / loop / info.py: 4%
887 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:55 -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].rstrip()
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. The new --visibility flag is the canonical interface;
189 # the legacy --all / --internal / --examples flags remain supported for backwards
190 # compatibility and map onto the same underlying tiers.
191 visibility_flag: str | None = getattr(args, "visibility", None)
192 show_all = getattr(args, "all", False) or visibility_flag == "all"
193 show_internal = getattr(args, "internal", False) or visibility_flag == "internal"
194 show_examples = getattr(args, "examples", False) or visibility_flag == "example"
195 # When --visibility public is explicit, enforce it even if other legacy flags are set.
196 show_public_only = visibility_flag == "public"
197 hidden_counts: dict[str, int] = {"internal": 0, "example": 0}
198 if not show_all:
199 if show_public_only:
200 for lp in all_loops:
201 vis = lp.get("visibility", "public")
202 if vis in hidden_counts:
203 hidden_counts[vis] += 1
204 all_loops = [lp for lp in all_loops if lp.get("visibility", "public") == "public"]
205 elif show_internal or show_examples:
206 wanted = set()
207 if show_internal:
208 wanted.add("internal")
209 if show_examples:
210 wanted.add("example")
211 all_loops = [lp for lp in all_loops if lp.get("visibility", "public") in wanted]
212 else:
213 for lp in all_loops:
214 vis = lp.get("visibility", "public")
215 if vis in hidden_counts:
216 hidden_counts[vis] += 1
217 all_loops = [lp for lp in all_loops if lp.get("visibility", "public") == "public"]
219 if not all_loops:
220 if getattr(args, "json", False):
221 print_json([])
222 return 0
223 print("No loops match the given filters")
224 return 0
226 if getattr(args, "json", False):
227 items: list[dict[str, Any]] = []
228 for lp in all_loops:
229 item: dict[str, Any] = {
230 "name": lp["name"],
231 "path": str(lp["path"]),
232 "category": lp["category"],
233 "labels": lp["labels"],
234 "visibility": lp.get("visibility", "public"),
235 "description": lp["description"],
236 }
237 if lp["builtin"]:
238 item["built_in"] = True
239 items.append(item)
240 print_json(items)
241 return 0
243 # Human-readable: group by category
244 buckets: dict[str, list[dict[str, Any]]] = {}
245 for lp in all_loops:
246 cat = lp["category"] or "uncategorized"
247 if cat not in buckets:
248 buckets[cat] = []
249 buckets[cat].append(lp)
251 # Sort categories; "uncategorized" always last
252 sorted_cats = sorted(c for c in buckets if c != "uncategorized")
253 if "uncategorized" in buckets:
254 sorted_cats.append("uncategorized")
256 # Cap name column so outlier names don't crush the description budget.
257 _MAX_NAME_COL = 32
258 _MAX_LABELS = 2
259 max_name_len = max((len(lp["name"]) for lp in all_loops), default=0)
260 name_col = min(max_name_len, _MAX_NAME_COL) + 2
261 tw = terminal_width()
263 # Summary header
264 n_project = sum(1 for lp in all_loops if not lp["builtin"])
265 n_builtin = len(all_loops) - n_project
266 summary = f" {len(all_loops)} loops · {len(buckets)} categories"
267 if n_project:
268 summary += f" · {n_project} project, {n_builtin} built-in"
269 print(colorize(summary, "2"))
270 print()
272 cats_printed = False
273 for cat in sorted_cats:
274 group = buckets[cat]
275 if cats_printed:
276 print() # blank line between category groups
277 cats_printed = True
278 cat_title = cat.replace("-", " ").title()
279 print(colorize(f" ▸ {cat_title} ({len(group)})", "36;1"))
280 for lp in group:
281 # Name: project loops get bold cyan, built-in loops get dimmer cyan
282 name_color = "36" if lp["builtin"] else "36;1"
283 display_name = (
284 _truncate(lp["name"], _MAX_NAME_COL)
285 if len(lp["name"]) > _MAX_NAME_COL
286 else lp["name"]
287 )
288 name_str = colorize(display_name.ljust(name_col), name_color)
290 # Suffix: cap labels; mark only project loops (built-in is the default)
291 suffix_parts: list[str] = []
292 labels = lp["labels"] or []
293 visible_labels = labels[:_MAX_LABELS]
294 hidden_label_count = len(labels) - _MAX_LABELS
295 for label in visible_labels:
296 suffix_parts.append(colorize(f"[{label}]", "2"))
297 if hidden_label_count > 0:
298 suffix_parts.append(colorize(f"[+{hidden_label_count}]", "2"))
299 if not lp["builtin"]:
300 suffix_parts.append(colorize("●", "36;1"))
302 if suffix_parts:
303 suffix_raw = " " + " ".join(suffix_parts)
304 suffix_visible = len(strip_ansi(suffix_raw))
305 else:
306 suffix_raw = ""
307 suffix_visible = 0
309 # Available width for description: indent + name_col + " " + desc + suffix
310 avail = tw - 2 - name_col - 2 - suffix_visible
311 desc_text = lp["description"] or ""
312 if desc_text and avail < len(desc_text):
313 desc_text = _truncate(desc_text, max(avail, 20))
314 desc_str = f" {colorize(desc_text, '2')}" if desc_text else ""
316 print(f" {name_str}{desc_str}{suffix_raw}")
318 # Footer: surface hidden tiers and point users at the natural-language router
319 # rather than asking them to scan dozens of loops.
320 print()
321 if n_project:
322 print(colorize(" ● = project loop (overrides built-in)", "2"))
323 hidden_bits: list[str] = []
324 if hidden_counts.get("internal"):
325 hidden_bits.append(f"{hidden_counts['internal']} internal (--internal)")
326 if hidden_counts.get("example"):
327 hidden_bits.append(f"{hidden_counts['example']} example (--examples)")
328 if hidden_bits:
329 print(colorize(f" Hidden: {', '.join(hidden_bits)} · all with --all", "2"))
330 print(
331 colorize(
332 ' Not sure which loop? `ll-loop run loop-router --input goal="<what you want>"`',
333 "2",
334 )
335 )
336 return 0
339_EVENT_TYPE_WIDTH = 16 # width of "handoff_detected"
342def _truncate(text: str, max_len: int) -> str:
343 """Truncate text to max_len with ellipsis."""
344 if max_len < 1:
345 return ""
346 if len(text) <= max_len:
347 return text
348 return text[: max_len - 1] + "\u2026"
351def _format_history_event(
352 event: dict[str, Any], verbose: bool, width: int, full: bool = False
353) -> str | None:
354 """Format a single history event. Returns None to skip the event."""
355 raw_ts = event.get("ts", "")
356 try:
357 ts = datetime.fromisoformat(raw_ts).strftime("%H:%M:%S")
358 except (ValueError, TypeError):
359 ts = raw_ts[:8] if len(raw_ts) >= 8 else raw_ts.ljust(8)
361 event_type = event.get("event", "unknown")
363 if event_type == "action_output" and not verbose:
364 return None
366 ts_str = colorize(ts, "2")
367 etype_padded = event_type.ljust(_EVENT_TYPE_WIDTH)
368 etype_color = "0"
369 detail = ""
370 extra_lines: list[str] = []
372 # Indentation prefix for verbose sub-lines (aligns under event detail column)
373 _indent = " " * (8 + 2 + _EVENT_TYPE_WIDTH + 2)
375 if event_type == "loop_start":
376 etype_color = "1"
377 detail = event.get("loop", "")
379 elif event_type == "loop_complete":
380 etype_color = "1"
381 final_state = event.get("final_state", "")
382 iterations = event.get("iterations", "")
383 terminated_by = event.get("terminated_by", "")
384 detail = f"{final_state} {iterations} iter [{terminated_by}]"
385 if error := event.get("error"):
386 detail += f" {colorize(error, '31')}"
388 elif event_type == "loop_resume":
389 etype_color = "1"
390 from_state = event.get("from_state", "")
391 iteration = event.get("iteration", "")
392 detail = f"from={from_state} iter={iteration}"
394 elif event_type == "state_enter":
395 etype_color = "34"
396 state = event.get("state", "")
397 iteration = event.get("iteration", "")
398 detail = f"{colorize(state, '1')} (iter {iteration})"
400 elif event_type == "action_start":
401 action = event.get("action", "")
402 is_prompt = event.get("is_prompt", False)
403 kind_label = "prompt" if is_prompt else "shell"
404 kind_str = colorize(f"[{kind_label}]", "2")
405 first_line = (
406 next((ln.strip() for ln in action.splitlines() if ln.strip()), "")
407 if is_prompt
408 else action
409 )
410 avail = width - 8 - 2 - _EVENT_TYPE_WIDTH - 2 - len(kind_label) - 2 - 2
411 detail = f"{_truncate(first_line, max(avail, 20))} {kind_str}"
413 elif event_type == "action_output":
414 # Only reached in verbose mode
415 etype_color = "2"
416 detail = colorize("\u2502 " + event.get("line", ""), "2")
418 elif event_type == "action_complete":
419 exit_code = event.get("exit_code", 0)
420 duration_ms = event.get("duration_ms", 0)
421 if exit_code == 0:
422 etype_color = "2"
423 status_str = colorize("\u2713", "32")
424 else:
425 etype_color = "38;5;208"
426 status_str = colorize(f"\u2717 exit={exit_code}", "38;5;208")
427 detail = f"{status_str} {duration_ms}ms"
428 is_prompt = event.get("is_prompt", False)
429 session_jsonl = event.get("session_jsonl") if is_prompt else None
430 if session_jsonl:
431 session_display = session_jsonl if verbose else os.path.basename(session_jsonl)
432 detail += f" session={colorize(session_display, '2')}"
433 if verbose:
434 output_preview = event.get("output_preview", "")
435 if output_preview:
436 avail_w = width - len(_indent) - 2
437 preview_text = (
438 output_preview if full else _truncate(output_preview, max(avail_w, 40))
439 )
440 for preview_line in preview_text.splitlines()[:5]:
441 extra_lines.append(colorize(_indent + "\u2502 " + preview_line, "2"))
443 elif event_type == "evaluate":
444 verdict = event.get("verdict", "")
445 confidence = event.get("confidence", "")
446 reason = event.get("reason", "")
447 if verdict == "yes":
448 etype_color = "32"
449 verdict_str = colorize("\u2713 yes", "32")
450 else:
451 etype_color = "38;5;208"
452 verdict_str = colorize(f"\u2717 {verdict}", "38;5;208")
453 conf_part = f" confidence={confidence}" if confidence != "" else ""
454 avail = width - 8 - 2 - _EVENT_TYPE_WIDTH - 2 - len("\u2713 yes") - len(conf_part) - 2
455 reason_part = f" {_truncate(reason, max(avail, 20))}" if reason else ""
456 detail = f"{verdict_str}{conf_part}{reason_part}"
457 if verbose:
458 llm_model = event.get("llm_model", "")
459 llm_latency_ms = event.get("llm_latency_ms", "")
460 llm_prompt = event.get("llm_prompt", "")
461 llm_raw_output = event.get("llm_raw_output", "")
462 if llm_model or llm_prompt:
463 meta_parts = []
464 if llm_model:
465 meta_parts.append(f"model={llm_model}")
466 if llm_latency_ms != "":
467 meta_parts.append(f"latency={llm_latency_ms}ms")
468 meta_str = " ".join(meta_parts)
469 extra_lines.append(
470 colorize(_indent + colorize("LLM Call", "2") + " " + meta_str, "2")
471 )
472 avail_w = width - len(_indent) - len("Prompt: ") - 2
473 if llm_prompt:
474 prompt_text = llm_prompt if full else _truncate(llm_prompt, max(avail_w, 40))
475 extra_lines.append(colorize(_indent + "Prompt: " + prompt_text, "2"))
476 if llm_raw_output:
477 resp_text = (
478 llm_raw_output if full else _truncate(llm_raw_output, max(avail_w, 40))
479 )
480 extra_lines.append(colorize(_indent + "Response: " + resp_text, "2"))
482 elif event_type == "route":
483 etype_color = "2"
484 from_state = event.get("from", "")
485 to_state = event.get("to", "")
486 detail = f"{from_state} \u2192 {colorize(to_state, '34')}"
488 elif event_type == "handoff_detected":
489 etype_color = "33"
490 detail = f"state={event.get('state', '')} iter={event.get('iteration', '')}"
492 else:
493 details = {k: v for k, v in event.items() if k not in ("event", "ts")}
494 detail = " ".join(f"{k}={v}" for k, v in details.items())
496 etype_str = colorize(etype_padded, etype_color)
497 main_line = f"{ts_str} {etype_str} {detail}"
498 if extra_lines:
499 return "\n".join([main_line] + extra_lines)
500 return main_line
503def _format_duration(ms: int) -> str:
504 """Format milliseconds as a human-readable duration."""
505 if ms < 1000:
506 return f"{ms}ms"
507 s = ms // 1000
508 if s < 60:
509 return f"{s}s"
510 m, s = divmod(s, 60)
511 if m < 60:
512 return f"{m}m{s:02d}s"
513 h, m = divmod(m, 60)
514 return f"{h}h{m:02d}m{s:02d}s"
517def _list_archived_runs(loop_name: str, loops_dir: Path, as_json: bool) -> int:
518 """List archived runs for a loop."""
519 import json as _json
521 from little_loops.fsm.persistence import HISTORY_DIR, LoopState
523 history_base = loops_dir / HISTORY_DIR
524 if not history_base.exists():
525 print(f"No history for: {loop_name}")
526 return 0
528 # Flat layout: run dirs are <run_id>-<loop_name> directly under .history/
529 suffix = f"-{loop_name}"
530 runs: list[tuple[str, LoopState | None]] = []
531 for run_dir in sorted(history_base.iterdir(), key=lambda d: d.name, reverse=True):
532 if not run_dir.is_dir() or not run_dir.name.endswith(suffix):
533 continue
534 run_id = run_dir.name[: -len(suffix)]
535 state_file = run_dir / "state.json"
536 state: LoopState | None = None
537 if state_file.exists():
538 try:
539 data = _json.loads(state_file.read_text())
540 state = LoopState.from_dict(data)
541 except (ValueError, KeyError):
542 pass
543 runs.append((run_id, state))
545 if not runs:
546 print(f"No history for: {loop_name}")
547 return 0
549 if as_json:
550 print(
551 _json.dumps(
552 [
553 {
554 "run_id": rid,
555 "status": s.status if s else None,
556 "started_at": s.started_at if s else None,
557 "iterations": s.iteration if s else None,
558 "duration_ms": s.accumulated_ms if s else None,
559 }
560 for rid, s in runs
561 ],
562 indent=2,
563 )
564 )
565 return 0
567 status_colors = {
568 "completed": "\033[32m",
569 "failed": "\033[31m",
570 "interrupted": "\033[33m",
571 "awaiting_continuation": "\033[36m",
572 "timed_out": "\033[33m",
573 "running": "\033[34m",
574 }
575 reset = "\033[0m"
577 print(f"Archived runs for: {loop_name} ({len(runs)} total)")
578 print()
580 for run_id, state in runs:
581 if state is not None:
582 color = status_colors.get(state.status, "")
583 status_str = f"{color}{state.status}{reset}"
584 duration_str = _format_duration(state.accumulated_ms) if state.accumulated_ms else "?"
585 started = state.started_at[:19].replace("T", " ") if state.started_at else "?"
586 iters = f"{state.iteration} iters"
587 else:
588 status_str = "unknown"
589 duration_str = "?"
590 started = "?"
591 iters = "?"
592 print(f" {run_id} {status_str} {started} {iters} {duration_str}")
594 print()
595 print(f"To view events: ll-loop history {loop_name} <run-id>")
596 return 0
599def cmd_history(
600 loop_name: str,
601 run_id: str | None,
602 args: argparse.Namespace,
603 loops_dir: Path,
604) -> int:
605 """Show loop history.
607 Without run_id: lists all archived runs with status and duration.
608 With run_id: shows events for that specific archived run.
609 """
610 tail = getattr(args, "tail", 50)
611 full = getattr(args, "full", False)
612 verbose = getattr(args, "verbose", False) or full
613 as_json = getattr(args, "json", False)
615 if run_id is None:
616 return _list_archived_runs(loop_name, loops_dir, as_json)
618 # Show events for a specific archived run
619 from little_loops.fsm.persistence import get_archived_events
621 events = get_archived_events(loop_name, run_id, loops_dir)
623 if not events:
624 print(f"No events found for run {run_id} of loop {loop_name}")
625 return 1
627 w = terminal_width()
628 if not verbose:
629 events = [e for e in events if e.get("event") != "action_output"]
631 # Apply optional filters (before --tail slice)
632 event_filter = getattr(args, "event", None)
633 if event_filter:
634 events = [e for e in events if e.get("event") == event_filter]
636 state_filter = getattr(args, "state", None)
637 if state_filter:
638 events = [
639 e
640 for e in events
641 if (
642 e.get("state") == state_filter
643 or e.get("from") == state_filter
644 or e.get("to") == state_filter
645 )
646 ]
648 since_str = getattr(args, "since", None)
649 if since_str:
650 from datetime import timedelta
652 from little_loops.text_utils import parse_duration
654 cutoff = datetime.now() - timedelta(seconds=parse_duration(since_str))
655 events = [
656 e
657 for e in events
658 if datetime.fromisoformat(e["ts"].replace("Z", "+00:00")).replace(tzinfo=None) >= cutoff
659 ]
661 if as_json:
662 print_json(events[-tail:])
663 return 0
664 for event in events[-tail:]:
665 line = _format_history_event(event, verbose, w, full=full)
666 if line is not None:
667 print(line)
669 return 0
672def cmd_audit_meta(loop_name: str, args: argparse.Namespace, loops_dir: Path) -> int:
673 """Summarize meta-eval.jsonl agreement stats from all archived runs of a loop.
675 Reads meta-eval.jsonl from each archived run, computes:
676 - Total iterations with llm_structured evaluate events
677 - Agreement rate (agreed / total)
678 - Mean diff size (files_changed) per verdict
679 - Divergence flags:
680 - agreed=false streak >=3 → "LLM optimistic drift detected"
681 - agreed=true with files_changed==0 streak >=3 → "Trivial agreement detected"
683 Returns 0 if no flags triggered, 1 if any threshold crossed.
684 """
685 import json as _json
687 from little_loops.fsm.persistence import HISTORY_DIR
689 history_root = loops_dir / HISTORY_DIR
690 if not history_root.exists():
691 print(f"No history for: {loop_name}")
692 return 0
694 suffix = f"-{loop_name}"
695 all_entries: list[dict[str, Any]] = []
697 for run_dir in sorted(history_root.iterdir(), key=lambda d: d.name):
698 if not run_dir.is_dir() or not run_dir.name.endswith(suffix):
699 continue
700 meta_eval_file = run_dir / "meta-eval.jsonl"
701 if not meta_eval_file.exists():
702 continue
703 for line in meta_eval_file.read_text(encoding="utf-8").splitlines():
704 line = line.strip()
705 if line:
706 try:
707 all_entries.append(_json.loads(line))
708 except _json.JSONDecodeError:
709 pass
711 if not all_entries:
712 print(f"No meta-eval data for: {loop_name}")
713 return 0
715 total = len(all_entries)
716 agreed_count = sum(1 for e in all_entries if e.get("agreed") is True)
717 agreement_rate = agreed_count / total if total else 0.0
719 # Mean diff size per verdict
720 agreed_sizes = [
721 e.get("diff_stats", {}).get("files_changed", 0) or 0
722 for e in all_entries
723 if e.get("agreed") is True
724 ]
725 disagreed_sizes = [
726 e.get("diff_stats", {}).get("files_changed", 0) or 0
727 for e in all_entries
728 if e.get("agreed") is False
729 ]
730 mean_agreed = sum(agreed_sizes) / len(agreed_sizes) if agreed_sizes else 0.0
731 mean_disagreed = sum(disagreed_sizes) / len(disagreed_sizes) if disagreed_sizes else 0.0
733 # Detect streaks
734 flags: list[str] = []
735 optimistic_streak = 0
736 trivial_streak = 0
737 max_optimistic = 0
738 max_trivial = 0
740 for entry in all_entries:
741 agreed = entry.get("agreed")
742 files_changed = (entry.get("diff_stats") or {}).get("files_changed", 0) or 0
744 if agreed is False:
745 optimistic_streak += 1
746 trivial_streak = 0
747 elif agreed is True and files_changed == 0:
748 trivial_streak += 1
749 optimistic_streak = 0
750 else:
751 optimistic_streak = 0
752 trivial_streak = 0
754 max_optimistic = max(max_optimistic, optimistic_streak)
755 max_trivial = max(max_trivial, trivial_streak)
757 if max_optimistic >= 3:
758 flags.append(f"LLM optimistic drift detected (streak={max_optimistic})")
759 if max_trivial >= 3:
760 flags.append(f"Trivial agreement detected (streak={max_trivial})")
762 as_json = getattr(args, "json", False)
763 if as_json:
764 result = {
765 "loop": loop_name,
766 "total_entries": total,
767 "agreed_count": agreed_count,
768 "agreement_rate": round(agreement_rate, 4),
769 "mean_files_changed_when_agreed": round(mean_agreed, 2),
770 "mean_files_changed_when_disagreed": round(mean_disagreed, 2),
771 "max_optimistic_streak": max_optimistic,
772 "max_trivial_streak": max_trivial,
773 "flags": flags,
774 }
775 print_json(result)
776 else:
777 print(f"Meta-eval audit: {loop_name}")
778 print(f" Total entries: {total}")
779 print(f" Agreed: {agreed_count}/{total} ({agreement_rate:.0%})")
780 print(f" Mean Δfiles agreed: {mean_agreed:.1f}")
781 print(f" Mean Δfiles disagreed: {mean_disagreed:.1f}")
782 if flags:
783 print()
784 for flag in flags:
785 print(f" ⚠ {flag}")
786 else:
787 print(" No divergence flags.")
789 if as_json:
790 return 0 # let caller inspect JSON for flags; non-zero kills provider.runCli
791 return 1 if flags else 0
794def cmd_diagnose_evaluators(loop_name: str, args: argparse.Namespace, loops_dir: Path) -> int:
795 """Detect non-discriminating evaluators from run history.
797 Walks .loops/.history/*-{loop_name}/events.jsonl, computes per-state
798 Bernoulli variance p*(1-p) on verdicts, flags states below threshold
799 with pattern-matched recommendations.
801 Returns 0 if no states flagged, 1 if any low-variance evaluators found.
802 """
803 from little_loops.analytics.variance import compute_evaluator_variance
805 threshold = getattr(args, "threshold", 0.05)
806 min_runs = getattr(args, "min_runs", 10)
807 as_json = getattr(args, "json", False)
809 report = compute_evaluator_variance(loop_name, loops_dir, threshold, min_runs)
811 if report is None:
812 if not (loops_dir / ".history").exists():
813 print(f"No history for: {loop_name}")
814 elif len(list((loops_dir / ".history").glob(f"*-{loop_name}"))) < min_runs:
815 runs_found = len(list((loops_dir / ".history").glob(f"*-{loop_name}")))
816 print(
817 f"Insufficient history for: {loop_name} "
818 f"(found {runs_found} run(s), need {min_runs})"
819 )
820 else:
821 print(f"No evaluate events for: {loop_name}")
822 return 0
824 flagged = [s for s in report.states if s.variance < threshold]
826 if as_json:
827 print_json(report.to_dict())
828 else:
829 print(f"Evaluator Variance Report (n={report.total_runs} runs)")
830 if not report.states:
831 print(" No evaluator states found in run history.")
832 for state in report.states:
833 disc_label = "✓ discriminating" if state.variance >= threshold else "⚠ low variance"
834 ci_str = ""
835 if state.ci is not None:
836 ci_str = f" CI=[{state.ci[0]:.2f}, {state.ci[1]:.2f}]"
837 print(
838 f" {state.state:20s} pass_rate={state.pass_rate:.2f} "
839 f"variance={state.variance:.2f}{ci_str} {disc_label}"
840 )
841 if state.recommendation:
842 for line in state.recommendation.split("\n"):
843 print(f" {line.strip()}")
845 if as_json:
846 return 0
847 return 1 if flagged else 0
850def cmd_calibrate_budget(loop_name: str, args: argparse.Namespace, loops_dir: Path) -> int:
851 """Report per-evaluator Bernoulli variance to guide max_steps calibration.
853 Calls compute_evaluator_variance() from analytics.variance and reports
854 p*(1-p) per evaluator state. Variance below threshold signals that
855 increasing max_steps is unlikely to help; fix the evaluator first.
857 Returns 0 if all evaluators are healthy, 1 if any are flagged.
858 """
859 from little_loops.analytics.variance import compute_evaluator_variance
861 threshold = getattr(args, "threshold", 0.05)
862 min_runs = getattr(args, "min_runs", 10)
863 as_json = getattr(args, "json", False)
865 report = compute_evaluator_variance(loop_name, loops_dir, threshold, min_runs)
867 if report is None:
868 if not (loops_dir / ".history").exists():
869 print(f"No history for: {loop_name}")
870 elif len(list((loops_dir / ".history").glob(f"*-{loop_name}"))) < min_runs:
871 runs_found = len(list((loops_dir / ".history").glob(f"*-{loop_name}")))
872 print(
873 f"Insufficient history for: {loop_name} "
874 f"(found {runs_found} run(s), need {min_runs})"
875 )
876 else:
877 print(f"No evaluate events for: {loop_name}")
878 return 0
880 flagged = [s for s in report.states if s.variance < threshold]
882 if as_json:
883 print_json(report.to_dict())
884 return 0
886 print(f"Loop: {loop_name}")
887 for state in report.states:
888 evaluator_type = state.evaluator_type or "unknown"
889 print(f"Evaluator: {state.state} ({evaluator_type})")
890 ci_str = ""
891 if state.ci is not None:
892 ci_str = f" CI=[{state.ci[0]:.2f}, {state.ci[1]:.2f}]"
893 if state.variance < threshold:
894 print(
895 f" Variance p*(1-p): {state.variance:.2f}{ci_str}"
896 f" ⚠ WARN: below {threshold} threshold"
897 f" — fix evaluator before increasing max_steps"
898 )
899 else:
900 print(f" Variance p*(1-p): {state.variance:.2f}{ci_str} ✓ OK")
902 return 1 if flagged else 0
905def cmd_promote_baseline(loop_name: str, args: argparse.Namespace, loops_dir: Path) -> int:
906 """Promote the latest run's action output as the new comparator baseline.
908 Reads action_output events from the most recent .history entry and writes
909 the concatenated output to .loops/baselines/<loop>/output.txt.
910 """
911 import json as _json
913 history_base = loops_dir / ".history"
914 if not history_base.exists():
915 print(f"No history for: {loop_name}")
916 return 1
918 suffix = f"-{loop_name}"
919 matched = sorted(
920 [d for d in history_base.iterdir() if d.is_dir() and d.name.endswith(suffix)],
921 key=lambda d: d.name,
922 reverse=True,
923 )
924 if not matched:
925 print(f"No history for: {loop_name}")
926 return 1
928 latest = matched[0]
929 events_file = latest / "events.jsonl"
930 if not events_file.exists():
931 print(f"No events.jsonl in latest run: {latest.name}")
932 return 1
934 lines = []
935 with open(events_file) as f:
936 for raw in f:
937 raw = raw.strip()
938 if not raw:
939 continue
940 try:
941 event = _json.loads(raw)
942 except _json.JSONDecodeError:
943 continue
944 if event.get("type") == "action_output":
945 line = event.get("line", "")
946 if line:
947 lines.append(line)
949 if not lines:
950 print(f"No action_output events in latest run: {latest.name}")
951 return 1
953 target = Path(loops_dir) / "baselines" / loop_name / "output.txt"
954 target.parent.mkdir(parents=True, exist_ok=True)
955 target.write_text("\n".join(lines))
956 print(f"Promoted baseline for {loop_name}: {target}")
957 return 0
960# ---------------------------------------------------------------------------
961# FSM diagram renderer — delegated to layout module (re-exported above)
962# ---------------------------------------------------------------------------
965# ---------------------------------------------------------------------------
966# State overview table
967# ---------------------------------------------------------------------------
970def _compact_transitions(state: StateConfig) -> str:
971 """Return a compact transition string for the overview table."""
972 raw: list[tuple[str, str]] = []
973 for label, target in [
974 ("yes", state.on_yes),
975 ("no", state.on_no),
976 ("error", state.on_error),
977 ("partial", state.on_partial),
978 ("next", state.next),
979 ]:
980 if target:
981 raw.append((label, target))
982 if state.route:
983 for verdict, target in state.route.routes.items():
984 raw.append((verdict, target))
985 if state.route.default:
986 raw.append(("_", state.route.default))
987 if not raw:
988 return "\u2014"
989 # Group by target, preserving first-seen order
990 seen: list[str] = []
991 by_target: dict[str, list[str]] = {}
992 for label, target in raw:
993 if target not in by_target:
994 seen.append(target)
995 by_target[target] = []
996 by_target[target].append(label)
997 return ", ".join(f"{'/'.join(by_target[t])}\u2192{t}" for t in seen)
1000def _print_state_overview_table(fsm: FSMLoop) -> None:
1001 """Print a compact summary table of all states."""
1002 rows: list[tuple[str, str, str, str]] = []
1003 for name, state in fsm.states.items():
1004 # State name column
1005 state_col = f"\u2192 {name}" if name == fsm.initial else f" {name}"
1007 # Type column
1008 if state.terminal:
1009 type_col = "\u2014"
1010 elif state.action_type:
1011 type_col = state.action_type
1012 elif state.action:
1013 type_col = "shell"
1014 else:
1015 type_col = "\u2014"
1017 # Action preview column
1018 if state.terminal:
1019 action_col = "(terminal)"
1020 elif state.action:
1021 src_lines = [ln.rstrip() for ln in state.action.strip().splitlines() if ln.rstrip()]
1022 action_col = src_lines[0] if src_lines else "\u2014"
1023 else:
1024 action_col = "\u2014"
1026 # Transitions column
1027 trans_col = _compact_transitions(state)
1028 rows.append((state_col, type_col, action_col, trans_col))
1030 if not rows:
1031 return
1033 tw = terminal_width()
1034 headers = ("State", "Type", "Action Preview", "Transitions")
1035 col0_w = max(len(headers[0]), max(len(r[0]) for r in rows))
1036 col1_w = max(len(headers[1]), max(len(r[1]) for r in rows))
1037 # Remaining width split between action preview and transitions
1038 fixed = col0_w + col1_w + 10 # margins + separators
1039 remaining = max(20, tw - fixed)
1040 col2_w = min(50, max(len(headers[2]), max(len(r[2]) for r in rows)), remaining * 3 // 5)
1041 col2_w = max(10, col2_w)
1042 col3_w = max(10, remaining - col2_w)
1044 print(f" {headers[0]:<{col0_w}} {headers[1]:<{col1_w}} {headers[2]:<{col2_w}} {headers[3]}")
1045 dash = "\u2500"
1046 print(f" {dash * col0_w} {dash * col1_w} {dash * col2_w} {dash * col3_w}")
1047 for state_col, type_col, action_col, trans_col in rows:
1048 if len(action_col) > col2_w:
1049 action_col = action_col[: col2_w - 1] + "\u2026"
1050 if len(trans_col) > col3_w:
1051 trans_col = trans_col[: col3_w - 1] + "\u2026"
1052 colored_type = colorize(type_col, "2") if type_col == "\u2014" else type_col
1053 print(
1054 f" {state_col:<{col0_w}} {colored_type:<{col1_w}} "
1055 f"{action_col:<{col2_w}} {trans_col}"
1056 )
1059# ---------------------------------------------------------------------------
1060# cmd_show
1061# ---------------------------------------------------------------------------
1063_EVALUATE_TYPE_DISPLAY: dict[str, str] = {
1064 "llm": "LLM",
1065 "llm_structured": "LLM (structured)",
1066 "exit_code": "exit code",
1067 "output_numeric": "numeric",
1068 "output_contains": "contains",
1069 "output_json": "JSON",
1070 "convergence": "convergence",
1071 "diff_stall": "diff stall",
1072 "action_stall": "action stall",
1073 "comparator": "blind comparator",
1074}
1077def _humanize_evaluate_type(ev_type: str) -> str:
1078 return _EVALUATE_TYPE_DISPLAY.get(ev_type, ev_type)
1081def cmd_show(
1082 loop_name: str,
1083 args: argparse.Namespace,
1084 loops_dir: Path,
1085 logger: Logger,
1086) -> int:
1087 """Show loop details and structure."""
1088 try:
1089 path = resolve_loop_path(loop_name, loops_dir)
1090 fsm, spec = load_loop_with_spec(loop_name, loops_dir, logger)
1091 except FileNotFoundError as e:
1092 logger.error(str(e))
1093 return 1
1094 except ValueError as e:
1095 logger.error(f"Invalid loop: {e}")
1096 return 1
1098 if getattr(args, "json", False) and getattr(args, "show_diagrams", None) is not None:
1099 logger.error("--json and --show-diagrams are mutually exclusive")
1100 return 1
1102 if getattr(args, "json", False):
1103 data = fsm.to_dict()
1104 if getattr(args, "resolved", False):
1105 for state in data.get("states", {}).values():
1106 if "loop" in state:
1107 try:
1108 child_path = resolve_loop_path(state["loop"], loops_dir)
1109 child_fsm, _ = load_and_validate(child_path)
1110 state["_subloop"] = child_fsm.to_dict().get("states", {})
1111 except (FileNotFoundError, ValueError):
1112 pass
1113 print_json(data)
1114 return 0
1116 tw = terminal_width()
1118 # Compute stats for header
1119 n_states = len(fsm.states)
1120 n_transitions = sum(
1121 bool(s.on_yes)
1122 + bool(s.on_no)
1123 + bool(s.on_error)
1124 + bool(s.on_partial)
1125 + bool(s.next)
1126 + bool(s.on_maintain)
1127 + (len(s.route.routes) + bool(s.route.default) if s.route else 0)
1128 for s in fsm.states.values()
1129 )
1131 # --- Compact metadata header ---
1132 # Line 1: ── name ───────── N states · M transitions ──
1133 stats_parts: list[str] = []
1134 stats_parts.append(f"{n_states} states")
1135 stats_parts.append(f"{n_transitions} transitions")
1136 stats_str = " \u00b7 ".join(stats_parts)
1138 header_left = f"\u2500\u2500 {loop_name} "
1139 header_right = f" {stats_str} \u2500\u2500"
1140 dashes = "\u2500" * max(0, tw - len(header_left) - len(header_right))
1141 print(f"{header_left}{dashes}{header_right}")
1143 # Line 2: source · max: N iter · handoff: X [· optional fields]
1144 config_parts: list[str] = [str(path), f"max: {fsm.max_steps} steps"]
1145 config_parts.append(f"handoff: {fsm.on_handoff}")
1146 if fsm.max_iterations is not None:
1147 config_parts.append(f"max_iterations: {fsm.max_iterations}")
1148 if fsm.on_max_iterations is not None:
1149 config_parts.append(f"on_max_iterations: {fsm.on_max_iterations}")
1150 if fsm.timeout:
1151 config_parts.append(f"timeout: {fsm.timeout}s")
1152 if fsm.backoff:
1153 config_parts.append(f"backoff: {fsm.backoff}s")
1154 if fsm.maintain:
1155 config_parts.append("maintain: yes")
1156 if fsm.context:
1157 config_parts.append(f"context: {', '.join(fsm.context.keys())}")
1158 if fsm.scope:
1159 config_parts.append(f"scope: {', '.join(fsm.scope)}")
1160 llm = fsm.llm
1161 llm_parts = []
1162 if llm.model != "sonnet":
1163 llm_parts.append(f"model={llm.model}")
1164 if llm.max_tokens != 256:
1165 llm_parts.append(f"max_tokens={llm.max_tokens}")
1166 if llm.timeout != 30:
1167 llm_parts.append(f"timeout={llm.timeout}s")
1168 if llm_parts:
1169 config_parts.append(f"llm: {', '.join(llm_parts)}")
1170 if fsm.config is not None:
1171 cfg_parts = []
1172 if fsm.config.handoff_threshold is not None:
1173 cfg_parts.append(f"handoff_threshold={fsm.config.handoff_threshold}")
1174 if fsm.config.readiness_threshold is not None:
1175 cfg_parts.append(f"readiness_threshold={fsm.config.readiness_threshold}")
1176 if fsm.config.outcome_threshold is not None:
1177 cfg_parts.append(f"outcome_threshold={fsm.config.outcome_threshold}")
1178 if fsm.config.max_continuations is not None:
1179 cfg_parts.append(f"max_continuations={fsm.config.max_continuations}")
1180 if cfg_parts:
1181 config_parts.append(f"config: {', '.join(cfg_parts)}")
1182 imports = spec.get("import", [])
1183 if imports:
1184 config_parts.append(f"imports: {', '.join(imports)}")
1185 print(" " + " \u00b7 ".join(config_parts))
1187 # --- Description ---
1188 description = spec.get("description", "").strip()
1189 if description:
1190 print()
1191 print("Description:")
1192 for line in description.splitlines():
1193 print(f" {line}")
1195 # --- ASCII FSM Diagram ---
1196 verbose = getattr(args, "verbose", False)
1197 from pathlib import Path
1199 from little_loops.config import BRConfig
1201 badges = BRConfig(Path.cwd()).loops.glyphs.to_dict()
1202 facets = resolve_facets(args)
1203 print()
1204 print("Diagram:")
1205 if facets is None:
1206 diagram = _render_fsm_diagram(fsm, verbose=verbose, badges=badges)
1207 else:
1208 diagram = _render_fsm_diagram(
1209 fsm,
1210 badges=badges,
1211 mode=facets.scope,
1212 suppress_labels=not facets.edge_labels,
1213 title_only=facets.state_detail == "title",
1214 )
1215 if diagram:
1216 print(diagram)
1218 # --- State overview table ---
1219 print()
1220 _print_state_overview_table(fsm)
1222 # --- States & Transitions (verbose only) ---
1223 if verbose:
1224 print()
1225 print("States:")
1226 first_state = True
1227 for name, state in fsm.states.items():
1228 if not first_state:
1229 print()
1230 first_state = False
1232 # Improved state section header: ── name ──── MARKERS · type ──
1233 right_parts = []
1234 if name == fsm.initial:
1235 right_parts.append("INITIAL")
1236 if state.terminal:
1237 right_parts.append("TERMINAL")
1238 if state.action_type:
1239 right_parts.append(state.action_type)
1240 right_info = " \u00b7 ".join(right_parts)
1241 inner_left = f"\u2500\u2500 {name} "
1242 inner_right = f" {right_info} \u2500\u2500" if right_info else " \u2500\u2500"
1243 fill = "\u2500" * max(0, tw - 2 - len(inner_left) - len(inner_right))
1244 print(f" {inner_left}{fill}{inner_right}")
1246 if state.action:
1247 if verbose:
1248 indented = "\n ".join(state.action.strip().splitlines())
1249 print(f" action:\n {indented}")
1250 elif state.action_type == "prompt":
1251 lines_act = state.action.strip().splitlines()
1252 preview = "\n ".join(lines_act[:3])
1253 if len(lines_act) > 3 or len(state.action) > 200:
1254 preview += " ..."
1255 print(f" action:\n {preview}")
1256 else: # shell, slash_command, or None
1257 action_display = (
1258 state.action[:70] + "..." if len(state.action) > 70 else state.action
1259 )
1260 print(f" action: {action_display}")
1261 if state.evaluate:
1262 ev = state.evaluate
1263 print(f" evaluate: {_humanize_evaluate_type(ev.type)}")
1264 if ev.prompt:
1265 if verbose:
1266 print(" prompt:")
1267 for line in ev.prompt.strip().splitlines():
1268 print(f" \u2502 {line}")
1269 else:
1270 ev_lines = ev.prompt.strip().splitlines()
1271 preview = ev_lines[0][:100] + (
1272 " ..." if len(ev_lines) > 1 or len(ev_lines[0]) > 100 else ""
1273 )
1274 print(f" prompt: {preview}")
1275 if ev.min_confidence != 0.5:
1276 print(f" min_confidence: {ev.min_confidence}")
1277 if ev.operator:
1278 print(f" operator: {ev.operator} {ev.target}")
1279 if ev.pattern:
1280 print(f" pattern: {ev.pattern}")
1281 if state.capture:
1282 print(f" capture: {state.capture}")
1283 if state.timeout:
1284 print(f" timeout: {state.timeout}s")
1285 # Collect (label, target) pairs
1286 raw_transitions: list[tuple[str, str]] = []
1287 for label, target in [
1288 ("yes", state.on_yes),
1289 ("no", state.on_no),
1290 ("error", state.on_error),
1291 ("partial", state.on_partial),
1292 ("next", state.next),
1293 ("maintain", state.on_maintain),
1294 ]:
1295 if target:
1296 raw_transitions.append((label, target))
1297 if state.route:
1298 for verdict, target in state.route.routes.items():
1299 raw_transitions.append((verdict, target))
1300 if state.route.default:
1301 raw_transitions.append(("_", state.route.default))
1302 # Group by target, preserving first-seen order
1303 target_labels: dict[str, list[str]] = {}
1304 seen_targets: list[str] = []
1305 for label, target in raw_transitions:
1306 if target not in target_labels:
1307 target_labels[target] = []
1308 seen_targets.append(target)
1309 target_labels[target].append(label)
1310 transitions = [
1311 f"{_colorize_label('/'.join(target_labels[t]))} \u2500\u2500\u2192 {t}"
1312 for t in seen_targets
1313 ]
1314 if transitions:
1315 print(" Transitions:")
1316 for t in transitions:
1317 print(f" {t}")
1319 # --- Commands ---
1320 print()
1321 print("Commands:")
1322 if fsm.commands:
1323 cmds = [(e.cmd, e.comment) for e in fsm.commands]
1324 else:
1325 cmds = [
1326 (f"ll-loop run {loop_name}", "run"),
1327 (f"ll-loop test {loop_name}", "single test iteration"),
1328 (f"ll-loop stop {loop_name}", "stop a running loop"),
1329 (f"ll-loop status {loop_name}", "check if running"),
1330 (f"ll-loop history {loop_name}", "execution history"),
1331 ]
1332 col_width = max(len(c) for c, _ in cmds) + 2
1333 for cmd, comment in cmds:
1334 print(f" {cmd:<{col_width}} # {comment}")
1336 return 0
1339def cmd_fragments(
1340 lib_path: str,
1341 args: argparse.Namespace,
1342 loops_dir: Path,
1343 logger: Logger,
1344) -> int:
1345 """List fragments in a library file with their descriptions.
1347 Resolves the library path relative to loops_dir or the built-in loops directory,
1348 then prints a table of fragment names and their description fields.
1349 """
1350 import yaml
1352 # Resolve path: absolute/direct → loops_dir-relative → builtin
1353 p = Path(lib_path)
1354 if not p.exists():
1355 candidate = loops_dir / lib_path
1356 if candidate.exists():
1357 p = candidate
1358 else:
1359 builtin_candidate = get_builtin_loops_dir() / lib_path
1360 if builtin_candidate.exists():
1361 p = builtin_candidate
1362 else:
1363 logger.error(f"Fragment library not found: {lib_path}")
1364 return 1
1366 try:
1367 with open(p) as f:
1368 data = yaml.safe_load(f) or {}
1369 except Exception as e:
1370 logger.error(f"Failed to load library: {e}")
1371 return 1
1373 fragments: dict[str, Any] = data.get("fragments", {})
1374 if not fragments:
1375 print(f"No fragments defined in: {p}")
1376 return 0
1378 tw = terminal_width()
1379 print(colorize(f"Fragments in {p.name} ({len(fragments)}):", "1"))
1380 print()
1382 max_name_len = max(len(name) for name in fragments)
1383 name_col_w = max(max_name_len, 8)
1384 desc_col_w = max(20, tw - name_col_w - 6)
1386 header_name = "Fragment".ljust(name_col_w)
1387 print(f" {colorize(header_name, '1')} Description")
1388 print(f" {'─' * name_col_w} {'─' * desc_col_w}")
1390 for name, frag in fragments.items():
1391 raw_desc = frag.get("description", "") if isinstance(frag, dict) else ""
1392 raw_desc = raw_desc.strip()
1393 first_line = raw_desc.splitlines()[0] if raw_desc else ""
1394 if first_line and len(first_line) > desc_col_w:
1395 first_line = first_line[: desc_col_w - 1] + "\u2026"
1396 desc_str = first_line if first_line else colorize("(no description)", "2")
1397 padded_name = name.ljust(name_col_w)
1398 print(f" {colorize(padded_name, '36;1')} {desc_str}")
1400 return 0