Coverage for little_loops / cli / loop / info.py: 4%
870 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-26 17:38 -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. 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 # Compute max name width for column alignment
257 max_name_len = max((len(lp["name"]) for lp in all_loops), default=0)
258 name_col = max_name_len + 2 # padding after longest name
259 tw = terminal_width()
261 cats_printed = False
262 for cat in sorted_cats:
263 group = buckets[cat]
264 if cats_printed:
265 print() # blank line between category groups
266 cats_printed = True
267 print(colorize(f"{cat} ({len(group)}):", "1"))
268 for lp in group:
269 # Name: project loops get bold cyan, built-in loops get dimmer cyan
270 name_color = "36" if lp["builtin"] else "36;1"
271 name_str = colorize(lp["name"].ljust(name_col), name_color)
273 # Suffix: labels + [built-in] tag
274 suffix_parts: list[str] = []
275 if lp["labels"]:
276 for label in lp["labels"]:
277 suffix_parts.append(colorize(f"[{label}]", "2"))
278 if lp["builtin"]:
279 suffix_parts.append(colorize("[built-in]", "2"))
281 if suffix_parts:
282 suffix_raw = " " + " ".join(suffix_parts)
283 suffix_visible = len(strip_ansi(suffix_raw))
284 else:
285 suffix_raw = ""
286 suffix_visible = 0
288 # Available width for description: indent + name_col + " " + desc + suffix
289 avail = tw - 2 - name_col - 2 - suffix_visible
290 desc_text = lp["description"] or ""
291 if desc_text and avail < len(desc_text):
292 desc_text = _truncate(desc_text, max(avail, 20))
293 desc_str = f" {colorize(desc_text, '2')}" if desc_text else ""
295 print(f" {name_str}{desc_str}{suffix_raw}")
297 # Footer: surface hidden tiers and point users at the natural-language router
298 # rather than asking them to scan dozens of loops.
299 hidden_bits: list[str] = []
300 if hidden_counts.get("internal"):
301 hidden_bits.append(f"{hidden_counts['internal']} internal (--internal)")
302 if hidden_counts.get("example"):
303 hidden_bits.append(f"{hidden_counts['example']} example (--examples)")
304 if hidden_bits:
305 print()
306 print(colorize(f"Hidden: {', '.join(hidden_bits)} · all with --all", "2"))
307 print(
308 colorize(
309 'Not sure which loop? `ll-loop run loop-router --input goal="<what you want>"`',
310 "2",
311 )
312 )
313 return 0
316_EVENT_TYPE_WIDTH = 16 # width of "handoff_detected"
319def _truncate(text: str, max_len: int) -> str:
320 """Truncate text to max_len with ellipsis."""
321 if max_len < 1:
322 return ""
323 if len(text) <= max_len:
324 return text
325 return text[: max_len - 1] + "\u2026"
328def _format_history_event(
329 event: dict[str, Any], verbose: bool, width: int, full: bool = False
330) -> str | None:
331 """Format a single history event. Returns None to skip the event."""
332 raw_ts = event.get("ts", "")
333 try:
334 ts = datetime.fromisoformat(raw_ts).strftime("%H:%M:%S")
335 except (ValueError, TypeError):
336 ts = raw_ts[:8] if len(raw_ts) >= 8 else raw_ts.ljust(8)
338 event_type = event.get("event", "unknown")
340 if event_type == "action_output" and not verbose:
341 return None
343 ts_str = colorize(ts, "2")
344 etype_padded = event_type.ljust(_EVENT_TYPE_WIDTH)
345 etype_color = "0"
346 detail = ""
347 extra_lines: list[str] = []
349 # Indentation prefix for verbose sub-lines (aligns under event detail column)
350 _indent = " " * (8 + 2 + _EVENT_TYPE_WIDTH + 2)
352 if event_type == "loop_start":
353 etype_color = "1"
354 detail = event.get("loop", "")
356 elif event_type == "loop_complete":
357 etype_color = "1"
358 final_state = event.get("final_state", "")
359 iterations = event.get("iterations", "")
360 terminated_by = event.get("terminated_by", "")
361 detail = f"{final_state} {iterations} iter [{terminated_by}]"
362 if error := event.get("error"):
363 detail += f" {colorize(error, '31')}"
365 elif event_type == "loop_resume":
366 etype_color = "1"
367 from_state = event.get("from_state", "")
368 iteration = event.get("iteration", "")
369 detail = f"from={from_state} iter={iteration}"
371 elif event_type == "state_enter":
372 etype_color = "34"
373 state = event.get("state", "")
374 iteration = event.get("iteration", "")
375 detail = f"{colorize(state, '1')} (iter {iteration})"
377 elif event_type == "action_start":
378 action = event.get("action", "")
379 is_prompt = event.get("is_prompt", False)
380 kind_label = "prompt" if is_prompt else "shell"
381 kind_str = colorize(f"[{kind_label}]", "2")
382 first_line = (
383 next((ln.strip() for ln in action.splitlines() if ln.strip()), "")
384 if is_prompt
385 else action
386 )
387 avail = width - 8 - 2 - _EVENT_TYPE_WIDTH - 2 - len(kind_label) - 2 - 2
388 detail = f"{_truncate(first_line, max(avail, 20))} {kind_str}"
390 elif event_type == "action_output":
391 # Only reached in verbose mode
392 etype_color = "2"
393 detail = colorize("\u2502 " + event.get("line", ""), "2")
395 elif event_type == "action_complete":
396 exit_code = event.get("exit_code", 0)
397 duration_ms = event.get("duration_ms", 0)
398 if exit_code == 0:
399 etype_color = "2"
400 status_str = colorize("\u2713", "32")
401 else:
402 etype_color = "38;5;208"
403 status_str = colorize(f"\u2717 exit={exit_code}", "38;5;208")
404 detail = f"{status_str} {duration_ms}ms"
405 is_prompt = event.get("is_prompt", False)
406 session_jsonl = event.get("session_jsonl") if is_prompt else None
407 if session_jsonl:
408 session_display = session_jsonl if verbose else os.path.basename(session_jsonl)
409 detail += f" session={colorize(session_display, '2')}"
410 if verbose:
411 output_preview = event.get("output_preview", "")
412 if output_preview:
413 avail_w = width - len(_indent) - 2
414 preview_text = (
415 output_preview if full else _truncate(output_preview, max(avail_w, 40))
416 )
417 for preview_line in preview_text.splitlines()[:5]:
418 extra_lines.append(colorize(_indent + "\u2502 " + preview_line, "2"))
420 elif event_type == "evaluate":
421 verdict = event.get("verdict", "")
422 confidence = event.get("confidence", "")
423 reason = event.get("reason", "")
424 if verdict == "yes":
425 etype_color = "32"
426 verdict_str = colorize("\u2713 yes", "32")
427 else:
428 etype_color = "38;5;208"
429 verdict_str = colorize(f"\u2717 {verdict}", "38;5;208")
430 conf_part = f" confidence={confidence}" if confidence != "" else ""
431 avail = width - 8 - 2 - _EVENT_TYPE_WIDTH - 2 - len("\u2713 yes") - len(conf_part) - 2
432 reason_part = f" {_truncate(reason, max(avail, 20))}" if reason else ""
433 detail = f"{verdict_str}{conf_part}{reason_part}"
434 if verbose:
435 llm_model = event.get("llm_model", "")
436 llm_latency_ms = event.get("llm_latency_ms", "")
437 llm_prompt = event.get("llm_prompt", "")
438 llm_raw_output = event.get("llm_raw_output", "")
439 if llm_model or llm_prompt:
440 meta_parts = []
441 if llm_model:
442 meta_parts.append(f"model={llm_model}")
443 if llm_latency_ms != "":
444 meta_parts.append(f"latency={llm_latency_ms}ms")
445 meta_str = " ".join(meta_parts)
446 extra_lines.append(
447 colorize(_indent + colorize("LLM Call", "2") + " " + meta_str, "2")
448 )
449 avail_w = width - len(_indent) - len("Prompt: ") - 2
450 if llm_prompt:
451 prompt_text = llm_prompt if full else _truncate(llm_prompt, max(avail_w, 40))
452 extra_lines.append(colorize(_indent + "Prompt: " + prompt_text, "2"))
453 if llm_raw_output:
454 resp_text = (
455 llm_raw_output if full else _truncate(llm_raw_output, max(avail_w, 40))
456 )
457 extra_lines.append(colorize(_indent + "Response: " + resp_text, "2"))
459 elif event_type == "route":
460 etype_color = "2"
461 from_state = event.get("from", "")
462 to_state = event.get("to", "")
463 detail = f"{from_state} \u2192 {colorize(to_state, '34')}"
465 elif event_type == "handoff_detected":
466 etype_color = "33"
467 detail = f"state={event.get('state', '')} iter={event.get('iteration', '')}"
469 else:
470 details = {k: v for k, v in event.items() if k not in ("event", "ts")}
471 detail = " ".join(f"{k}={v}" for k, v in details.items())
473 etype_str = colorize(etype_padded, etype_color)
474 main_line = f"{ts_str} {etype_str} {detail}"
475 if extra_lines:
476 return "\n".join([main_line] + extra_lines)
477 return main_line
480def _format_duration(ms: int) -> str:
481 """Format milliseconds as a human-readable duration."""
482 if ms < 1000:
483 return f"{ms}ms"
484 s = ms // 1000
485 if s < 60:
486 return f"{s}s"
487 m, s = divmod(s, 60)
488 if m < 60:
489 return f"{m}m{s:02d}s"
490 h, m = divmod(m, 60)
491 return f"{h}h{m:02d}m{s:02d}s"
494def _list_archived_runs(loop_name: str, loops_dir: Path, as_json: bool) -> int:
495 """List archived runs for a loop."""
496 import json as _json
498 from little_loops.fsm.persistence import HISTORY_DIR, LoopState
500 history_base = loops_dir / HISTORY_DIR
501 if not history_base.exists():
502 print(f"No history for: {loop_name}")
503 return 0
505 # Flat layout: run dirs are <run_id>-<loop_name> directly under .history/
506 suffix = f"-{loop_name}"
507 runs: list[tuple[str, LoopState | None]] = []
508 for run_dir in sorted(history_base.iterdir(), key=lambda d: d.name, reverse=True):
509 if not run_dir.is_dir() or not run_dir.name.endswith(suffix):
510 continue
511 run_id = run_dir.name[: -len(suffix)]
512 state_file = run_dir / "state.json"
513 state: LoopState | None = None
514 if state_file.exists():
515 try:
516 data = _json.loads(state_file.read_text())
517 state = LoopState.from_dict(data)
518 except (ValueError, KeyError):
519 pass
520 runs.append((run_id, state))
522 if not runs:
523 print(f"No history for: {loop_name}")
524 return 0
526 if as_json:
527 print(
528 _json.dumps(
529 [
530 {
531 "run_id": rid,
532 "status": s.status if s else None,
533 "started_at": s.started_at if s else None,
534 "iterations": s.iteration if s else None,
535 "duration_ms": s.accumulated_ms if s else None,
536 }
537 for rid, s in runs
538 ],
539 indent=2,
540 )
541 )
542 return 0
544 status_colors = {
545 "completed": "\033[32m",
546 "failed": "\033[31m",
547 "interrupted": "\033[33m",
548 "awaiting_continuation": "\033[36m",
549 "timed_out": "\033[33m",
550 "running": "\033[34m",
551 }
552 reset = "\033[0m"
554 print(f"Archived runs for: {loop_name} ({len(runs)} total)")
555 print()
557 for run_id, state in runs:
558 if state is not None:
559 color = status_colors.get(state.status, "")
560 status_str = f"{color}{state.status}{reset}"
561 duration_str = _format_duration(state.accumulated_ms) if state.accumulated_ms else "?"
562 started = state.started_at[:19].replace("T", " ") if state.started_at else "?"
563 iters = f"{state.iteration} iters"
564 else:
565 status_str = "unknown"
566 duration_str = "?"
567 started = "?"
568 iters = "?"
569 print(f" {run_id} {status_str} {started} {iters} {duration_str}")
571 print()
572 print(f"To view events: ll-loop history {loop_name} <run-id>")
573 return 0
576def cmd_history(
577 loop_name: str,
578 run_id: str | None,
579 args: argparse.Namespace,
580 loops_dir: Path,
581) -> int:
582 """Show loop history.
584 Without run_id: lists all archived runs with status and duration.
585 With run_id: shows events for that specific archived run.
586 """
587 tail = getattr(args, "tail", 50)
588 full = getattr(args, "full", False)
589 verbose = getattr(args, "verbose", False) or full
590 as_json = getattr(args, "json", False)
592 if run_id is None:
593 return _list_archived_runs(loop_name, loops_dir, as_json)
595 # Show events for a specific archived run
596 from little_loops.fsm.persistence import get_archived_events
598 events = get_archived_events(loop_name, run_id, loops_dir)
600 if not events:
601 print(f"No events found for run {run_id} of loop {loop_name}")
602 return 1
604 w = terminal_width()
605 if not verbose:
606 events = [e for e in events if e.get("event") != "action_output"]
608 # Apply optional filters (before --tail slice)
609 event_filter = getattr(args, "event", None)
610 if event_filter:
611 events = [e for e in events if e.get("event") == event_filter]
613 state_filter = getattr(args, "state", None)
614 if state_filter:
615 events = [
616 e
617 for e in events
618 if (
619 e.get("state") == state_filter
620 or e.get("from") == state_filter
621 or e.get("to") == state_filter
622 )
623 ]
625 since_str = getattr(args, "since", None)
626 if since_str:
627 from datetime import timedelta
629 from little_loops.text_utils import parse_duration
631 cutoff = datetime.now() - timedelta(seconds=parse_duration(since_str))
632 events = [
633 e
634 for e in events
635 if datetime.fromisoformat(e["ts"].replace("Z", "+00:00")).replace(tzinfo=None) >= cutoff
636 ]
638 if as_json:
639 print_json(events[-tail:])
640 return 0
641 for event in events[-tail:]:
642 line = _format_history_event(event, verbose, w, full=full)
643 if line is not None:
644 print(line)
646 return 0
649def cmd_audit_meta(loop_name: str, args: argparse.Namespace, loops_dir: Path) -> int:
650 """Summarize meta-eval.jsonl agreement stats from all archived runs of a loop.
652 Reads meta-eval.jsonl from each archived run, computes:
653 - Total iterations with llm_structured evaluate events
654 - Agreement rate (agreed / total)
655 - Mean diff size (files_changed) per verdict
656 - Divergence flags:
657 - agreed=false streak >=3 → "LLM optimistic drift detected"
658 - agreed=true with files_changed==0 streak >=3 → "Trivial agreement detected"
660 Returns 0 if no flags triggered, 1 if any threshold crossed.
661 """
662 import json as _json
664 from little_loops.fsm.persistence import HISTORY_DIR
666 history_root = loops_dir / HISTORY_DIR
667 if not history_root.exists():
668 print(f"No history for: {loop_name}")
669 return 0
671 suffix = f"-{loop_name}"
672 all_entries: list[dict[str, Any]] = []
674 for run_dir in sorted(history_root.iterdir(), key=lambda d: d.name):
675 if not run_dir.is_dir() or not run_dir.name.endswith(suffix):
676 continue
677 meta_eval_file = run_dir / "meta-eval.jsonl"
678 if not meta_eval_file.exists():
679 continue
680 for line in meta_eval_file.read_text(encoding="utf-8").splitlines():
681 line = line.strip()
682 if line:
683 try:
684 all_entries.append(_json.loads(line))
685 except _json.JSONDecodeError:
686 pass
688 if not all_entries:
689 print(f"No meta-eval data for: {loop_name}")
690 return 0
692 total = len(all_entries)
693 agreed_count = sum(1 for e in all_entries if e.get("agreed") is True)
694 agreement_rate = agreed_count / total if total else 0.0
696 # Mean diff size per verdict
697 agreed_sizes = [
698 e.get("diff_stats", {}).get("files_changed", 0) or 0
699 for e in all_entries
700 if e.get("agreed") is True
701 ]
702 disagreed_sizes = [
703 e.get("diff_stats", {}).get("files_changed", 0) or 0
704 for e in all_entries
705 if e.get("agreed") is False
706 ]
707 mean_agreed = sum(agreed_sizes) / len(agreed_sizes) if agreed_sizes else 0.0
708 mean_disagreed = sum(disagreed_sizes) / len(disagreed_sizes) if disagreed_sizes else 0.0
710 # Detect streaks
711 flags: list[str] = []
712 optimistic_streak = 0
713 trivial_streak = 0
714 max_optimistic = 0
715 max_trivial = 0
717 for entry in all_entries:
718 agreed = entry.get("agreed")
719 files_changed = (entry.get("diff_stats") or {}).get("files_changed", 0) or 0
721 if agreed is False:
722 optimistic_streak += 1
723 trivial_streak = 0
724 elif agreed is True and files_changed == 0:
725 trivial_streak += 1
726 optimistic_streak = 0
727 else:
728 optimistic_streak = 0
729 trivial_streak = 0
731 max_optimistic = max(max_optimistic, optimistic_streak)
732 max_trivial = max(max_trivial, trivial_streak)
734 if max_optimistic >= 3:
735 flags.append(f"LLM optimistic drift detected (streak={max_optimistic})")
736 if max_trivial >= 3:
737 flags.append(f"Trivial agreement detected (streak={max_trivial})")
739 as_json = getattr(args, "json", False)
740 if as_json:
741 result = {
742 "loop": loop_name,
743 "total_entries": total,
744 "agreed_count": agreed_count,
745 "agreement_rate": round(agreement_rate, 4),
746 "mean_files_changed_when_agreed": round(mean_agreed, 2),
747 "mean_files_changed_when_disagreed": round(mean_disagreed, 2),
748 "max_optimistic_streak": max_optimistic,
749 "max_trivial_streak": max_trivial,
750 "flags": flags,
751 }
752 print_json(result)
753 else:
754 print(f"Meta-eval audit: {loop_name}")
755 print(f" Total entries: {total}")
756 print(f" Agreed: {agreed_count}/{total} ({agreement_rate:.0%})")
757 print(f" Mean Δfiles agreed: {mean_agreed:.1f}")
758 print(f" Mean Δfiles disagreed: {mean_disagreed:.1f}")
759 if flags:
760 print()
761 for flag in flags:
762 print(f" ⚠ {flag}")
763 else:
764 print(" No divergence flags.")
766 if as_json:
767 return 0 # let caller inspect JSON for flags; non-zero kills provider.runCli
768 return 1 if flags else 0
771def cmd_diagnose_evaluators(loop_name: str, args: argparse.Namespace, loops_dir: Path) -> int:
772 """Detect non-discriminating evaluators from run history.
774 Walks .loops/.history/*-{loop_name}/events.jsonl, computes per-state
775 Bernoulli variance p*(1-p) on verdicts, flags states below threshold
776 with pattern-matched recommendations.
778 Returns 0 if no states flagged, 1 if any low-variance evaluators found.
779 """
780 from little_loops.analytics.variance import compute_evaluator_variance
782 threshold = getattr(args, "threshold", 0.05)
783 min_runs = getattr(args, "min_runs", 10)
784 as_json = getattr(args, "json", False)
786 report = compute_evaluator_variance(loop_name, loops_dir, threshold, min_runs)
788 if report is None:
789 if not (loops_dir / ".history").exists():
790 print(f"No history for: {loop_name}")
791 elif len(list((loops_dir / ".history").glob(f"*-{loop_name}"))) < min_runs:
792 runs_found = len(list((loops_dir / ".history").glob(f"*-{loop_name}")))
793 print(
794 f"Insufficient history for: {loop_name} "
795 f"(found {runs_found} run(s), need {min_runs})"
796 )
797 else:
798 print(f"No evaluate events for: {loop_name}")
799 return 0
801 flagged = [s for s in report.states if s.variance < threshold]
803 if as_json:
804 print_json(report.to_dict())
805 else:
806 print(f"Evaluator Variance Report (n={report.total_runs} runs)")
807 if not report.states:
808 print(" No evaluator states found in run history.")
809 for state in report.states:
810 disc_label = "✓ discriminating" if state.variance >= threshold else "⚠ low variance"
811 ci_str = ""
812 if state.ci is not None:
813 ci_str = f" CI=[{state.ci[0]:.2f}, {state.ci[1]:.2f}]"
814 print(
815 f" {state.state:20s} pass_rate={state.pass_rate:.2f} "
816 f"variance={state.variance:.2f}{ci_str} {disc_label}"
817 )
818 if state.recommendation:
819 for line in state.recommendation.split("\n"):
820 print(f" {line.strip()}")
822 if as_json:
823 return 0
824 return 1 if flagged else 0
827def cmd_calibrate_budget(loop_name: str, args: argparse.Namespace, loops_dir: Path) -> int:
828 """Report per-evaluator Bernoulli variance to guide max_steps calibration.
830 Calls compute_evaluator_variance() from analytics.variance and reports
831 p*(1-p) per evaluator state. Variance below threshold signals that
832 increasing max_steps is unlikely to help; fix the evaluator first.
834 Returns 0 if all evaluators are healthy, 1 if any are flagged.
835 """
836 from little_loops.analytics.variance import compute_evaluator_variance
838 threshold = getattr(args, "threshold", 0.05)
839 min_runs = getattr(args, "min_runs", 10)
840 as_json = getattr(args, "json", False)
842 report = compute_evaluator_variance(loop_name, loops_dir, threshold, min_runs)
844 if report is None:
845 if not (loops_dir / ".history").exists():
846 print(f"No history for: {loop_name}")
847 elif len(list((loops_dir / ".history").glob(f"*-{loop_name}"))) < min_runs:
848 runs_found = len(list((loops_dir / ".history").glob(f"*-{loop_name}")))
849 print(
850 f"Insufficient history for: {loop_name} "
851 f"(found {runs_found} run(s), need {min_runs})"
852 )
853 else:
854 print(f"No evaluate events for: {loop_name}")
855 return 0
857 flagged = [s for s in report.states if s.variance < threshold]
859 if as_json:
860 print_json(report.to_dict())
861 return 0
863 print(f"Loop: {loop_name}")
864 for state in report.states:
865 evaluator_type = state.evaluator_type or "unknown"
866 print(f"Evaluator: {state.state} ({evaluator_type})")
867 ci_str = ""
868 if state.ci is not None:
869 ci_str = f" CI=[{state.ci[0]:.2f}, {state.ci[1]:.2f}]"
870 if state.variance < threshold:
871 print(
872 f" Variance p*(1-p): {state.variance:.2f}{ci_str}"
873 f" ⚠ WARN: below {threshold} threshold"
874 f" — fix evaluator before increasing max_steps"
875 )
876 else:
877 print(f" Variance p*(1-p): {state.variance:.2f}{ci_str} ✓ OK")
879 return 1 if flagged else 0
882def cmd_promote_baseline(loop_name: str, args: argparse.Namespace, loops_dir: Path) -> int:
883 """Promote the latest run's action output as the new comparator baseline.
885 Reads action_output events from the most recent .history entry and writes
886 the concatenated output to .loops/baselines/<loop>/output.txt.
887 """
888 import json as _json
890 history_base = loops_dir / ".history"
891 if not history_base.exists():
892 print(f"No history for: {loop_name}")
893 return 1
895 suffix = f"-{loop_name}"
896 matched = sorted(
897 [d for d in history_base.iterdir() if d.is_dir() and d.name.endswith(suffix)],
898 key=lambda d: d.name,
899 reverse=True,
900 )
901 if not matched:
902 print(f"No history for: {loop_name}")
903 return 1
905 latest = matched[0]
906 events_file = latest / "events.jsonl"
907 if not events_file.exists():
908 print(f"No events.jsonl in latest run: {latest.name}")
909 return 1
911 lines = []
912 with open(events_file) as f:
913 for raw in f:
914 raw = raw.strip()
915 if not raw:
916 continue
917 try:
918 event = _json.loads(raw)
919 except _json.JSONDecodeError:
920 continue
921 if event.get("type") == "action_output":
922 line = event.get("line", "")
923 if line:
924 lines.append(line)
926 if not lines:
927 print(f"No action_output events in latest run: {latest.name}")
928 return 1
930 target = Path(loops_dir) / "baselines" / loop_name / "output.txt"
931 target.parent.mkdir(parents=True, exist_ok=True)
932 target.write_text("\n".join(lines))
933 print(f"Promoted baseline for {loop_name}: {target}")
934 return 0
937# ---------------------------------------------------------------------------
938# FSM diagram renderer — delegated to layout module (re-exported above)
939# ---------------------------------------------------------------------------
942# ---------------------------------------------------------------------------
943# State overview table
944# ---------------------------------------------------------------------------
947def _compact_transitions(state: StateConfig) -> str:
948 """Return a compact transition string for the overview table."""
949 raw: list[tuple[str, str]] = []
950 for label, target in [
951 ("yes", state.on_yes),
952 ("no", state.on_no),
953 ("error", state.on_error),
954 ("partial", state.on_partial),
955 ("next", state.next),
956 ]:
957 if target:
958 raw.append((label, target))
959 if state.route:
960 for verdict, target in state.route.routes.items():
961 raw.append((verdict, target))
962 if state.route.default:
963 raw.append(("_", state.route.default))
964 if not raw:
965 return "\u2014"
966 # Group by target, preserving first-seen order
967 seen: list[str] = []
968 by_target: dict[str, list[str]] = {}
969 for label, target in raw:
970 if target not in by_target:
971 seen.append(target)
972 by_target[target] = []
973 by_target[target].append(label)
974 return ", ".join(f"{'/'.join(by_target[t])}\u2192{t}" for t in seen)
977def _print_state_overview_table(fsm: FSMLoop) -> None:
978 """Print a compact summary table of all states."""
979 rows: list[tuple[str, str, str, str]] = []
980 for name, state in fsm.states.items():
981 # State name column
982 state_col = f"\u2192 {name}" if name == fsm.initial else f" {name}"
984 # Type column
985 if state.terminal:
986 type_col = "\u2014"
987 elif state.action_type:
988 type_col = state.action_type
989 elif state.action:
990 type_col = "shell"
991 else:
992 type_col = "\u2014"
994 # Action preview column
995 if state.terminal:
996 action_col = "(terminal)"
997 elif state.action:
998 src_lines = [ln.rstrip() for ln in state.action.strip().splitlines() if ln.rstrip()]
999 action_col = src_lines[0] if src_lines else "\u2014"
1000 else:
1001 action_col = "\u2014"
1003 # Transitions column
1004 trans_col = _compact_transitions(state)
1005 rows.append((state_col, type_col, action_col, trans_col))
1007 if not rows:
1008 return
1010 tw = terminal_width()
1011 headers = ("State", "Type", "Action Preview", "Transitions")
1012 col0_w = max(len(headers[0]), max(len(r[0]) for r in rows))
1013 col1_w = max(len(headers[1]), max(len(r[1]) for r in rows))
1014 # Remaining width split between action preview and transitions
1015 fixed = col0_w + col1_w + 10 # margins + separators
1016 remaining = max(20, tw - fixed)
1017 col2_w = min(50, max(len(headers[2]), max(len(r[2]) for r in rows)), remaining * 3 // 5)
1018 col2_w = max(10, col2_w)
1019 col3_w = max(10, remaining - col2_w)
1021 print(f" {headers[0]:<{col0_w}} {headers[1]:<{col1_w}} {headers[2]:<{col2_w}} {headers[3]}")
1022 dash = "\u2500"
1023 print(f" {dash * col0_w} {dash * col1_w} {dash * col2_w} {dash * col3_w}")
1024 for state_col, type_col, action_col, trans_col in rows:
1025 if len(action_col) > col2_w:
1026 action_col = action_col[: col2_w - 1] + "\u2026"
1027 if len(trans_col) > col3_w:
1028 trans_col = trans_col[: col3_w - 1] + "\u2026"
1029 colored_type = colorize(type_col, "2") if type_col == "\u2014" else type_col
1030 print(
1031 f" {state_col:<{col0_w}} {colored_type:<{col1_w}} "
1032 f"{action_col:<{col2_w}} {trans_col}"
1033 )
1036# ---------------------------------------------------------------------------
1037# cmd_show
1038# ---------------------------------------------------------------------------
1040_EVALUATE_TYPE_DISPLAY: dict[str, str] = {
1041 "llm": "LLM",
1042 "llm_structured": "LLM (structured)",
1043 "exit_code": "exit code",
1044 "output_numeric": "numeric",
1045 "output_contains": "contains",
1046 "output_json": "JSON",
1047 "convergence": "convergence",
1048 "diff_stall": "diff stall",
1049 "action_stall": "action stall",
1050 "comparator": "blind comparator",
1051}
1054def _humanize_evaluate_type(ev_type: str) -> str:
1055 return _EVALUATE_TYPE_DISPLAY.get(ev_type, ev_type)
1058def cmd_show(
1059 loop_name: str,
1060 args: argparse.Namespace,
1061 loops_dir: Path,
1062 logger: Logger,
1063) -> int:
1064 """Show loop details and structure."""
1065 try:
1066 path = resolve_loop_path(loop_name, loops_dir)
1067 fsm, spec = load_loop_with_spec(loop_name, loops_dir, logger)
1068 except FileNotFoundError as e:
1069 logger.error(str(e))
1070 return 1
1071 except ValueError as e:
1072 logger.error(f"Invalid loop: {e}")
1073 return 1
1075 if getattr(args, "json", False) and getattr(args, "show_diagrams", None) is not None:
1076 logger.error("--json and --show-diagrams are mutually exclusive")
1077 return 1
1079 if getattr(args, "json", False):
1080 data = fsm.to_dict()
1081 if getattr(args, "resolved", False):
1082 for state in data.get("states", {}).values():
1083 if "loop" in state:
1084 try:
1085 child_path = resolve_loop_path(state["loop"], loops_dir)
1086 child_fsm, _ = load_and_validate(child_path)
1087 state["_subloop"] = child_fsm.to_dict().get("states", {})
1088 except (FileNotFoundError, ValueError):
1089 pass
1090 print_json(data)
1091 return 0
1093 tw = terminal_width()
1095 # Compute stats for header
1096 n_states = len(fsm.states)
1097 n_transitions = sum(
1098 bool(s.on_yes)
1099 + bool(s.on_no)
1100 + bool(s.on_error)
1101 + bool(s.on_partial)
1102 + bool(s.next)
1103 + bool(s.on_maintain)
1104 + (len(s.route.routes) + bool(s.route.default) if s.route else 0)
1105 for s in fsm.states.values()
1106 )
1108 # --- Compact metadata header ---
1109 # Line 1: ── name ───────── N states · M transitions ──
1110 stats_parts: list[str] = []
1111 stats_parts.append(f"{n_states} states")
1112 stats_parts.append(f"{n_transitions} transitions")
1113 stats_str = " \u00b7 ".join(stats_parts)
1115 header_left = f"\u2500\u2500 {loop_name} "
1116 header_right = f" {stats_str} \u2500\u2500"
1117 dashes = "\u2500" * max(0, tw - len(header_left) - len(header_right))
1118 print(f"{header_left}{dashes}{header_right}")
1120 # Line 2: source · max: N iter · handoff: X [· optional fields]
1121 config_parts: list[str] = [str(path), f"max: {fsm.max_steps} steps"]
1122 config_parts.append(f"handoff: {fsm.on_handoff}")
1123 if fsm.max_iterations is not None:
1124 config_parts.append(f"max_iterations: {fsm.max_iterations}")
1125 if fsm.on_max_iterations is not None:
1126 config_parts.append(f"on_max_iterations: {fsm.on_max_iterations}")
1127 if fsm.timeout:
1128 config_parts.append(f"timeout: {fsm.timeout}s")
1129 if fsm.backoff:
1130 config_parts.append(f"backoff: {fsm.backoff}s")
1131 if fsm.maintain:
1132 config_parts.append("maintain: yes")
1133 if fsm.context:
1134 config_parts.append(f"context: {', '.join(fsm.context.keys())}")
1135 if fsm.scope:
1136 config_parts.append(f"scope: {', '.join(fsm.scope)}")
1137 llm = fsm.llm
1138 llm_parts = []
1139 if llm.model != "sonnet":
1140 llm_parts.append(f"model={llm.model}")
1141 if llm.max_tokens != 256:
1142 llm_parts.append(f"max_tokens={llm.max_tokens}")
1143 if llm.timeout != 30:
1144 llm_parts.append(f"timeout={llm.timeout}s")
1145 if llm_parts:
1146 config_parts.append(f"llm: {', '.join(llm_parts)}")
1147 if fsm.config is not None:
1148 cfg_parts = []
1149 if fsm.config.handoff_threshold is not None:
1150 cfg_parts.append(f"handoff_threshold={fsm.config.handoff_threshold}")
1151 if fsm.config.readiness_threshold is not None:
1152 cfg_parts.append(f"readiness_threshold={fsm.config.readiness_threshold}")
1153 if fsm.config.outcome_threshold is not None:
1154 cfg_parts.append(f"outcome_threshold={fsm.config.outcome_threshold}")
1155 if fsm.config.max_continuations is not None:
1156 cfg_parts.append(f"max_continuations={fsm.config.max_continuations}")
1157 if cfg_parts:
1158 config_parts.append(f"config: {', '.join(cfg_parts)}")
1159 imports = spec.get("import", [])
1160 if imports:
1161 config_parts.append(f"imports: {', '.join(imports)}")
1162 print(" " + " \u00b7 ".join(config_parts))
1164 # --- Description ---
1165 description = spec.get("description", "").strip()
1166 if description:
1167 print()
1168 print("Description:")
1169 for line in description.splitlines():
1170 print(f" {line}")
1172 # --- ASCII FSM Diagram ---
1173 verbose = getattr(args, "verbose", False)
1174 from pathlib import Path
1176 from little_loops.config import BRConfig
1178 badges = BRConfig(Path.cwd()).loops.glyphs.to_dict()
1179 facets = resolve_facets(args)
1180 print()
1181 print("Diagram:")
1182 if facets is None:
1183 diagram = _render_fsm_diagram(fsm, verbose=verbose, badges=badges)
1184 else:
1185 diagram = _render_fsm_diagram(
1186 fsm,
1187 badges=badges,
1188 mode=facets.scope,
1189 suppress_labels=not facets.edge_labels,
1190 title_only=facets.state_detail == "title",
1191 )
1192 if diagram:
1193 print(diagram)
1195 # --- State overview table ---
1196 print()
1197 _print_state_overview_table(fsm)
1199 # --- States & Transitions (verbose only) ---
1200 if verbose:
1201 print()
1202 print("States:")
1203 first_state = True
1204 for name, state in fsm.states.items():
1205 if not first_state:
1206 print()
1207 first_state = False
1209 # Improved state section header: ── name ──── MARKERS · type ──
1210 right_parts = []
1211 if name == fsm.initial:
1212 right_parts.append("INITIAL")
1213 if state.terminal:
1214 right_parts.append("TERMINAL")
1215 if state.action_type:
1216 right_parts.append(state.action_type)
1217 right_info = " \u00b7 ".join(right_parts)
1218 inner_left = f"\u2500\u2500 {name} "
1219 inner_right = f" {right_info} \u2500\u2500" if right_info else " \u2500\u2500"
1220 fill = "\u2500" * max(0, tw - 2 - len(inner_left) - len(inner_right))
1221 print(f" {inner_left}{fill}{inner_right}")
1223 if state.action:
1224 if verbose:
1225 indented = "\n ".join(state.action.strip().splitlines())
1226 print(f" action:\n {indented}")
1227 elif state.action_type == "prompt":
1228 lines_act = state.action.strip().splitlines()
1229 preview = "\n ".join(lines_act[:3])
1230 if len(lines_act) > 3 or len(state.action) > 200:
1231 preview += " ..."
1232 print(f" action:\n {preview}")
1233 else: # shell, slash_command, or None
1234 action_display = (
1235 state.action[:70] + "..." if len(state.action) > 70 else state.action
1236 )
1237 print(f" action: {action_display}")
1238 if state.evaluate:
1239 ev = state.evaluate
1240 print(f" evaluate: {_humanize_evaluate_type(ev.type)}")
1241 if ev.prompt:
1242 if verbose:
1243 print(" prompt:")
1244 for line in ev.prompt.strip().splitlines():
1245 print(f" \u2502 {line}")
1246 else:
1247 ev_lines = ev.prompt.strip().splitlines()
1248 preview = ev_lines[0][:100] + (
1249 " ..." if len(ev_lines) > 1 or len(ev_lines[0]) > 100 else ""
1250 )
1251 print(f" prompt: {preview}")
1252 if ev.min_confidence != 0.5:
1253 print(f" min_confidence: {ev.min_confidence}")
1254 if ev.operator:
1255 print(f" operator: {ev.operator} {ev.target}")
1256 if ev.pattern:
1257 print(f" pattern: {ev.pattern}")
1258 if state.capture:
1259 print(f" capture: {state.capture}")
1260 if state.timeout:
1261 print(f" timeout: {state.timeout}s")
1262 # Collect (label, target) pairs
1263 raw_transitions: list[tuple[str, str]] = []
1264 for label, target in [
1265 ("yes", state.on_yes),
1266 ("no", state.on_no),
1267 ("error", state.on_error),
1268 ("partial", state.on_partial),
1269 ("next", state.next),
1270 ("maintain", state.on_maintain),
1271 ]:
1272 if target:
1273 raw_transitions.append((label, target))
1274 if state.route:
1275 for verdict, target in state.route.routes.items():
1276 raw_transitions.append((verdict, target))
1277 if state.route.default:
1278 raw_transitions.append(("_", state.route.default))
1279 # Group by target, preserving first-seen order
1280 target_labels: dict[str, list[str]] = {}
1281 seen_targets: list[str] = []
1282 for label, target in raw_transitions:
1283 if target not in target_labels:
1284 target_labels[target] = []
1285 seen_targets.append(target)
1286 target_labels[target].append(label)
1287 transitions = [
1288 f"{_colorize_label('/'.join(target_labels[t]))} \u2500\u2500\u2192 {t}"
1289 for t in seen_targets
1290 ]
1291 if transitions:
1292 print(" Transitions:")
1293 for t in transitions:
1294 print(f" {t}")
1296 # --- Commands ---
1297 print()
1298 print("Commands:")
1299 if fsm.commands:
1300 cmds = [(e.cmd, e.comment) for e in fsm.commands]
1301 else:
1302 cmds = [
1303 (f"ll-loop run {loop_name}", "run"),
1304 (f"ll-loop test {loop_name}", "single test iteration"),
1305 (f"ll-loop stop {loop_name}", "stop a running loop"),
1306 (f"ll-loop status {loop_name}", "check if running"),
1307 (f"ll-loop history {loop_name}", "execution history"),
1308 ]
1309 col_width = max(len(c) for c, _ in cmds) + 2
1310 for cmd, comment in cmds:
1311 print(f" {cmd:<{col_width}} # {comment}")
1313 return 0
1316def cmd_fragments(
1317 lib_path: str,
1318 args: argparse.Namespace,
1319 loops_dir: Path,
1320 logger: Logger,
1321) -> int:
1322 """List fragments in a library file with their descriptions.
1324 Resolves the library path relative to loops_dir or the built-in loops directory,
1325 then prints a table of fragment names and their description fields.
1326 """
1327 import yaml
1329 # Resolve path: absolute/direct → loops_dir-relative → builtin
1330 p = Path(lib_path)
1331 if not p.exists():
1332 candidate = loops_dir / lib_path
1333 if candidate.exists():
1334 p = candidate
1335 else:
1336 builtin_candidate = get_builtin_loops_dir() / lib_path
1337 if builtin_candidate.exists():
1338 p = builtin_candidate
1339 else:
1340 logger.error(f"Fragment library not found: {lib_path}")
1341 return 1
1343 try:
1344 with open(p) as f:
1345 data = yaml.safe_load(f) or {}
1346 except Exception as e:
1347 logger.error(f"Failed to load library: {e}")
1348 return 1
1350 fragments: dict[str, Any] = data.get("fragments", {})
1351 if not fragments:
1352 print(f"No fragments defined in: {p}")
1353 return 0
1355 tw = terminal_width()
1356 print(colorize(f"Fragments in {p.name} ({len(fragments)}):", "1"))
1357 print()
1359 max_name_len = max(len(name) for name in fragments)
1360 name_col_w = max(max_name_len, 8)
1361 desc_col_w = max(20, tw - name_col_w - 6)
1363 header_name = "Fragment".ljust(name_col_w)
1364 print(f" {colorize(header_name, '1')} Description")
1365 print(f" {'─' * name_col_w} {'─' * desc_col_w}")
1367 for name, frag in fragments.items():
1368 raw_desc = frag.get("description", "") if isinstance(frag, dict) else ""
1369 raw_desc = raw_desc.strip()
1370 first_line = raw_desc.splitlines()[0] if raw_desc else ""
1371 if first_line and len(first_line) > desc_col_w:
1372 first_line = first_line[: desc_col_w - 1] + "\u2026"
1373 desc_str = first_line if first_line else colorize("(no description)", "2")
1374 padded_name = name.ljust(name_col_w)
1375 print(f" {colorize(padded_name, '36;1')} {desc_str}")
1377 return 0