Coverage for little_loops / cli / loop / info.py: 4%
792 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-04 12:21 -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.schema import FSMLoop, StateConfig
27from little_loops.fsm.validation import load_and_validate
28from little_loops.logger import Logger
31def _load_loop_meta(path: Path) -> dict[str, Any]:
32 """Return metadata from a loop YAML file (description, category, labels)."""
33 import yaml
35 try:
36 with open(path) as f:
37 spec = yaml.safe_load(f) or {}
38 desc_raw = spec.get("description", "") or ""
39 if desc_raw.strip():
40 raw_lines = desc_raw.splitlines()
41 desc = raw_lines[0]
42 if len(raw_lines) > 1:
43 desc += "…"
44 else:
45 desc = ""
46 category = spec.get("category", "") or ""
47 labels: list[str] = spec.get("labels", []) or []
48 return {"description": desc, "category": category, "labels": labels}
49 except Exception:
50 return {"description": "", "category": "", "labels": []}
53def cmd_list(
54 args: argparse.Namespace,
55 loops_dir: Path,
56) -> int:
57 """List loops."""
58 status_filter = getattr(args, "status", None)
59 if getattr(args, "running", False) or status_filter:
60 from little_loops.fsm.persistence import list_running_loops
62 states = list_running_loops(loops_dir)
63 if status_filter:
64 states = [s for s in states if s.status == status_filter]
65 if not states:
66 if status_filter:
67 print(f"No loops with status: {status_filter}")
68 return 1
69 print("No running loops")
70 return 0
71 if getattr(args, "json", False):
72 print_json([s.to_dict() for s in states])
73 return 0
74 print(colorize("Running loops:", "1"))
75 _STATUS_COLORS = {"running": "32", "interrupted": "33", "stopped": "2", "starting": "33"}
77 # Group by loop_name to avoid duplicate rows for multi-instance loops
78 from collections import defaultdict
80 groups: dict[str, list] = defaultdict(list)
81 for state in states:
82 groups[state.loop_name].append(state)
84 for loop_name_key, group_states in groups.items():
85 if len(group_states) == 1:
86 state = group_states[0]
87 elapsed_s = (state.accumulated_ms or 0) // 1000
88 elapsed_str = (
89 f"{elapsed_s}s" if elapsed_s < 60 else f"{elapsed_s // 60}m {elapsed_s % 60}s"
90 )
91 name_str = colorize(state.loop_name, "1")
92 state_str = colorize(state.current_state, "34")
93 status_color = _STATUS_COLORS.get(state.status, "2")
94 display_status = "paused" if state.status == "interrupted" else state.status
95 status_str = colorize(f"[{display_status}]", status_color)
96 elapsed_colored = colorize(elapsed_str, "2")
97 print(
98 f" {name_str}: {state_str} (iteration {state.iteration})"
99 f" {status_str} {elapsed_colored}"
100 )
101 else:
102 # Multiple instances: show a grouped summary
103 name_str = colorize(loop_name_key, "1")
104 statuses = ", ".join(
105 colorize(
106 f"[{'paused' if s.status == 'interrupted' else s.status}]",
107 _STATUS_COLORS.get(s.status, "2"),
108 )
109 for s in group_states
110 )
111 count_str = colorize(f"({len(group_states)} instances)", "2")
112 print(f" {name_str}: {count_str} {statuses}")
113 return 0
115 builtin_only = getattr(args, "builtin", False)
117 def _rel_key(path: Path, base: Path) -> str:
118 """Relative-path identifier matching what `ll-loop run` accepts."""
119 return str(path.relative_to(base).with_suffix(""))
121 # Collect project loops (skipped when --builtin is set). Recurse into
122 # subdirectories (e.g. oracles/) and filter to runnable FSM definitions so
123 # library fragments under loops/lib/ stay hidden.
124 project_names: set[str] = set()
125 yaml_files: list[Path] = []
126 if not builtin_only and loops_dir.exists():
127 yaml_files = sorted(p for p in loops_dir.rglob("*.yaml") if is_runnable_loop(p))
128 project_names = {_rel_key(p, loops_dir) for p in yaml_files}
130 # Collect built-in loops (excluding those overridden by a project loop at
131 # the same relative path).
132 builtin_dir = get_builtin_loops_dir()
133 builtin_files: list[Path] = []
134 if builtin_dir.exists():
135 builtin_files = [
136 f
137 for f in sorted(builtin_dir.rglob("*.yaml"))
138 if is_runnable_loop(f) and _rel_key(f, builtin_dir) not in project_names
139 ]
141 if not yaml_files and not builtin_files:
142 if getattr(args, "json", False):
143 print_json([])
144 return 0
145 print("No loops available")
146 return 0
148 # Build combined metadata list
149 all_loops: list[dict[str, Any]] = []
150 for path in yaml_files:
151 meta = _load_loop_meta(path)
152 all_loops.append(
153 {"name": _rel_key(path, loops_dir), "path": path, "builtin": False, **meta}
154 )
155 for path in builtin_files:
156 meta = _load_loop_meta(path)
157 all_loops.append(
158 {"name": _rel_key(path, builtin_dir), "path": path, "builtin": True, **meta}
159 )
161 # Apply --category filter
162 category_filter = getattr(args, "category", None)
163 if category_filter:
164 all_loops = [lp for lp in all_loops if lp["category"] == category_filter]
166 # Apply --label filter (action="append" → list or None)
167 label_filters: list[str] = getattr(args, "label", None) or []
168 if label_filters:
169 all_loops = [
170 lp
171 for lp in all_loops
172 if any(lf.lower() in [lb.lower() for lb in lp["labels"]] for lf in label_filters)
173 ]
175 if not all_loops:
176 if getattr(args, "json", False):
177 print_json([])
178 return 0
179 print("No loops match the given filters")
180 return 0
182 if getattr(args, "json", False):
183 items: list[dict[str, Any]] = []
184 for lp in all_loops:
185 item: dict[str, Any] = {
186 "name": lp["name"],
187 "path": str(lp["path"]),
188 "category": lp["category"],
189 "labels": lp["labels"],
190 "description": lp["description"],
191 }
192 if lp["builtin"]:
193 item["built_in"] = True
194 items.append(item)
195 print_json(items)
196 return 0
198 # Human-readable: group by category
199 buckets: dict[str, list[dict[str, Any]]] = {}
200 for lp in all_loops:
201 cat = lp["category"] or "uncategorized"
202 if cat not in buckets:
203 buckets[cat] = []
204 buckets[cat].append(lp)
206 # Sort categories; "uncategorized" always last
207 sorted_cats = sorted(c for c in buckets if c != "uncategorized")
208 if "uncategorized" in buckets:
209 sorted_cats.append("uncategorized")
211 # Compute max name width for column alignment
212 max_name_len = max((len(lp["name"]) for lp in all_loops), default=0)
213 name_col = max_name_len + 2 # padding after longest name
214 tw = terminal_width()
216 cats_printed = False
217 for cat in sorted_cats:
218 group = buckets[cat]
219 if cats_printed:
220 print() # blank line between category groups
221 cats_printed = True
222 print(colorize(f"{cat} ({len(group)}):", "1"))
223 for lp in group:
224 # Name: project loops get bold cyan, built-in loops get dimmer cyan
225 name_color = "36" if lp["builtin"] else "36;1"
226 name_str = colorize(lp["name"].ljust(name_col), name_color)
228 # Suffix: labels + [built-in] tag
229 suffix_parts: list[str] = []
230 if lp["labels"]:
231 for label in lp["labels"]:
232 suffix_parts.append(colorize(f"[{label}]", "2"))
233 if lp["builtin"]:
234 suffix_parts.append(colorize("[built-in]", "2"))
236 if suffix_parts:
237 suffix_raw = " " + " ".join(suffix_parts)
238 suffix_visible = len(strip_ansi(suffix_raw))
239 else:
240 suffix_raw = ""
241 suffix_visible = 0
243 # Available width for description: indent + name_col + " " + desc + suffix
244 avail = tw - 2 - name_col - 2 - suffix_visible
245 desc_text = lp["description"] or ""
246 if desc_text and avail < len(desc_text):
247 desc_text = _truncate(desc_text, max(avail, 20))
248 desc_str = f" {colorize(desc_text, '2')}" if desc_text else ""
250 print(f" {name_str}{desc_str}{suffix_raw}")
251 return 0
254_EVENT_TYPE_WIDTH = 16 # width of "handoff_detected"
257def _truncate(text: str, max_len: int) -> str:
258 """Truncate text to max_len with ellipsis."""
259 if max_len < 1:
260 return ""
261 if len(text) <= max_len:
262 return text
263 return text[: max_len - 1] + "\u2026"
266def _format_history_event(
267 event: dict[str, Any], verbose: bool, width: int, full: bool = False
268) -> str | None:
269 """Format a single history event. Returns None to skip the event."""
270 raw_ts = event.get("ts", "")
271 try:
272 ts = datetime.fromisoformat(raw_ts).strftime("%H:%M:%S")
273 except (ValueError, TypeError):
274 ts = raw_ts[:8] if len(raw_ts) >= 8 else raw_ts.ljust(8)
276 event_type = event.get("event", "unknown")
278 if event_type == "action_output" and not verbose:
279 return None
281 ts_str = colorize(ts, "2")
282 etype_padded = event_type.ljust(_EVENT_TYPE_WIDTH)
283 etype_color = "0"
284 detail = ""
285 extra_lines: list[str] = []
287 # Indentation prefix for verbose sub-lines (aligns under event detail column)
288 _indent = " " * (8 + 2 + _EVENT_TYPE_WIDTH + 2)
290 if event_type == "loop_start":
291 etype_color = "1"
292 detail = event.get("loop", "")
294 elif event_type == "loop_complete":
295 etype_color = "1"
296 final_state = event.get("final_state", "")
297 iterations = event.get("iterations", "")
298 terminated_by = event.get("terminated_by", "")
299 detail = f"{final_state} {iterations} iter [{terminated_by}]"
301 elif event_type == "loop_resume":
302 etype_color = "1"
303 from_state = event.get("from_state", "")
304 iteration = event.get("iteration", "")
305 detail = f"from={from_state} iter={iteration}"
307 elif event_type == "state_enter":
308 etype_color = "34"
309 state = event.get("state", "")
310 iteration = event.get("iteration", "")
311 detail = f"{colorize(state, '1')} (iter {iteration})"
313 elif event_type == "action_start":
314 action = event.get("action", "")
315 is_prompt = event.get("is_prompt", False)
316 kind_label = "prompt" if is_prompt else "shell"
317 kind_str = colorize(f"[{kind_label}]", "2")
318 first_line = (
319 next((ln.strip() for ln in action.splitlines() if ln.strip()), "")
320 if is_prompt
321 else action
322 )
323 avail = width - 8 - 2 - _EVENT_TYPE_WIDTH - 2 - len(kind_label) - 2 - 2
324 detail = f"{_truncate(first_line, max(avail, 20))} {kind_str}"
326 elif event_type == "action_output":
327 # Only reached in verbose mode
328 etype_color = "2"
329 detail = colorize("\u2502 " + event.get("line", ""), "2")
331 elif event_type == "action_complete":
332 exit_code = event.get("exit_code", 0)
333 duration_ms = event.get("duration_ms", 0)
334 if exit_code == 0:
335 etype_color = "2"
336 status_str = colorize("\u2713", "32")
337 else:
338 etype_color = "38;5;208"
339 status_str = colorize(f"\u2717 exit={exit_code}", "38;5;208")
340 detail = f"{status_str} {duration_ms}ms"
341 is_prompt = event.get("is_prompt", False)
342 session_jsonl = event.get("session_jsonl") if is_prompt else None
343 if session_jsonl:
344 session_display = session_jsonl if verbose else os.path.basename(session_jsonl)
345 detail += f" session={colorize(session_display, '2')}"
346 if verbose:
347 output_preview = event.get("output_preview", "")
348 if output_preview:
349 avail_w = width - len(_indent) - 2
350 preview_text = (
351 output_preview if full else _truncate(output_preview, max(avail_w, 40))
352 )
353 for preview_line in preview_text.splitlines()[:5]:
354 extra_lines.append(colorize(_indent + "\u2502 " + preview_line, "2"))
356 elif event_type == "evaluate":
357 verdict = event.get("verdict", "")
358 confidence = event.get("confidence", "")
359 reason = event.get("reason", "")
360 if verdict == "yes":
361 etype_color = "32"
362 verdict_str = colorize("\u2713 yes", "32")
363 else:
364 etype_color = "38;5;208"
365 verdict_str = colorize(f"\u2717 {verdict}", "38;5;208")
366 conf_part = f" confidence={confidence}" if confidence != "" else ""
367 avail = width - 8 - 2 - _EVENT_TYPE_WIDTH - 2 - len("\u2713 yes") - len(conf_part) - 2
368 reason_part = f" {_truncate(reason, max(avail, 20))}" if reason else ""
369 detail = f"{verdict_str}{conf_part}{reason_part}"
370 if verbose:
371 llm_model = event.get("llm_model", "")
372 llm_latency_ms = event.get("llm_latency_ms", "")
373 llm_prompt = event.get("llm_prompt", "")
374 llm_raw_output = event.get("llm_raw_output", "")
375 if llm_model or llm_prompt:
376 meta_parts = []
377 if llm_model:
378 meta_parts.append(f"model={llm_model}")
379 if llm_latency_ms != "":
380 meta_parts.append(f"latency={llm_latency_ms}ms")
381 meta_str = " ".join(meta_parts)
382 extra_lines.append(
383 colorize(_indent + colorize("LLM Call", "2") + " " + meta_str, "2")
384 )
385 avail_w = width - len(_indent) - len("Prompt: ") - 2
386 if llm_prompt:
387 prompt_text = llm_prompt if full else _truncate(llm_prompt, max(avail_w, 40))
388 extra_lines.append(colorize(_indent + "Prompt: " + prompt_text, "2"))
389 if llm_raw_output:
390 resp_text = (
391 llm_raw_output if full else _truncate(llm_raw_output, max(avail_w, 40))
392 )
393 extra_lines.append(colorize(_indent + "Response: " + resp_text, "2"))
395 elif event_type == "route":
396 etype_color = "2"
397 from_state = event.get("from", "")
398 to_state = event.get("to", "")
399 detail = f"{from_state} \u2192 {colorize(to_state, '34')}"
401 elif event_type == "handoff_detected":
402 etype_color = "33"
403 detail = f"state={event.get('state', '')} iter={event.get('iteration', '')}"
405 else:
406 details = {k: v for k, v in event.items() if k not in ("event", "ts")}
407 detail = " ".join(f"{k}={v}" for k, v in details.items())
409 etype_str = colorize(etype_padded, etype_color)
410 main_line = f"{ts_str} {etype_str} {detail}"
411 if extra_lines:
412 return "\n".join([main_line] + extra_lines)
413 return main_line
416def _format_duration(ms: int) -> str:
417 """Format milliseconds as a human-readable duration."""
418 if ms < 1000:
419 return f"{ms}ms"
420 s = ms // 1000
421 if s < 60:
422 return f"{s}s"
423 m, s = divmod(s, 60)
424 if m < 60:
425 return f"{m}m{s:02d}s"
426 h, m = divmod(m, 60)
427 return f"{h}h{m:02d}m{s:02d}s"
430def _list_archived_runs(loop_name: str, loops_dir: Path, as_json: bool) -> int:
431 """List archived runs for a loop."""
432 import json as _json
434 from little_loops.fsm.persistence import HISTORY_DIR, LoopState
436 history_base = loops_dir / HISTORY_DIR
437 if not history_base.exists():
438 print(f"No history for: {loop_name}")
439 return 0
441 # Flat layout: run dirs are <run_id>-<loop_name> directly under .history/
442 suffix = f"-{loop_name}"
443 runs: list[tuple[str, LoopState | None]] = []
444 for run_dir in sorted(history_base.iterdir(), key=lambda d: d.name, reverse=True):
445 if not run_dir.is_dir() or not run_dir.name.endswith(suffix):
446 continue
447 run_id = run_dir.name[: -len(suffix)]
448 state_file = run_dir / "state.json"
449 state: LoopState | None = None
450 if state_file.exists():
451 try:
452 data = _json.loads(state_file.read_text())
453 state = LoopState.from_dict(data)
454 except (ValueError, KeyError):
455 pass
456 runs.append((run_id, state))
458 if not runs:
459 print(f"No history for: {loop_name}")
460 return 0
462 if as_json:
463 print(
464 _json.dumps(
465 [
466 {
467 "run_id": rid,
468 "status": s.status if s else None,
469 "started_at": s.started_at if s else None,
470 "iterations": s.iteration if s else None,
471 "duration_ms": s.accumulated_ms if s else None,
472 }
473 for rid, s in runs
474 ],
475 indent=2,
476 )
477 )
478 return 0
480 status_colors = {
481 "completed": "\033[32m",
482 "failed": "\033[31m",
483 "interrupted": "\033[33m",
484 "awaiting_continuation": "\033[36m",
485 "timed_out": "\033[33m",
486 "running": "\033[34m",
487 }
488 reset = "\033[0m"
490 print(f"Archived runs for: {loop_name} ({len(runs)} total)")
491 print()
493 for run_id, state in runs:
494 if state is not None:
495 color = status_colors.get(state.status, "")
496 status_str = f"{color}{state.status}{reset}"
497 duration_str = _format_duration(state.accumulated_ms) if state.accumulated_ms else "?"
498 started = state.started_at[:19].replace("T", " ") if state.started_at else "?"
499 iters = f"{state.iteration} iters"
500 else:
501 status_str = "unknown"
502 duration_str = "?"
503 started = "?"
504 iters = "?"
505 print(f" {run_id} {status_str} {started} {iters} {duration_str}")
507 print()
508 print(f"To view events: ll-loop history {loop_name} <run-id>")
509 return 0
512def cmd_history(
513 loop_name: str,
514 run_id: str | None,
515 args: argparse.Namespace,
516 loops_dir: Path,
517) -> int:
518 """Show loop history.
520 Without run_id: lists all archived runs with status and duration.
521 With run_id: shows events for that specific archived run.
522 """
523 tail = getattr(args, "tail", 50)
524 full = getattr(args, "full", False)
525 verbose = getattr(args, "verbose", False) or full
526 as_json = getattr(args, "json", False)
528 if run_id is None:
529 return _list_archived_runs(loop_name, loops_dir, as_json)
531 # Show events for a specific archived run
532 from little_loops.fsm.persistence import get_archived_events
534 events = get_archived_events(loop_name, run_id, loops_dir)
536 if not events:
537 print(f"No events found for run {run_id} of loop {loop_name}")
538 return 1
540 w = terminal_width()
541 if not verbose:
542 events = [e for e in events if e.get("event") != "action_output"]
544 # Apply optional filters (before --tail slice)
545 event_filter = getattr(args, "event", None)
546 if event_filter:
547 events = [e for e in events if e.get("event") == event_filter]
549 state_filter = getattr(args, "state", None)
550 if state_filter:
551 events = [
552 e
553 for e in events
554 if (
555 e.get("state") == state_filter
556 or e.get("from") == state_filter
557 or e.get("to") == state_filter
558 )
559 ]
561 since_str = getattr(args, "since", None)
562 if since_str:
563 from datetime import timedelta
565 from little_loops.text_utils import parse_duration
567 cutoff = datetime.now() - timedelta(seconds=parse_duration(since_str))
568 events = [
569 e
570 for e in events
571 if datetime.fromisoformat(e["ts"].replace("Z", "+00:00")).replace(tzinfo=None) >= cutoff
572 ]
574 if as_json:
575 print_json(events[-tail:])
576 return 0
577 for event in events[-tail:]:
578 line = _format_history_event(event, verbose, w, full=full)
579 if line is not None:
580 print(line)
582 return 0
585def cmd_audit_meta(loop_name: str, args: argparse.Namespace, loops_dir: Path) -> int:
586 """Summarize meta-eval.jsonl agreement stats from all archived runs of a loop.
588 Reads meta-eval.jsonl from each archived run, computes:
589 - Total iterations with llm_structured evaluate events
590 - Agreement rate (agreed / total)
591 - Mean diff size (files_changed) per verdict
592 - Divergence flags:
593 - agreed=false streak >=3 → "LLM optimistic drift detected"
594 - agreed=true with files_changed==0 streak >=3 → "Trivial agreement detected"
596 Returns 0 if no flags triggered, 1 if any threshold crossed.
597 """
598 import json as _json
600 from little_loops.fsm.persistence import HISTORY_DIR
602 history_root = loops_dir / HISTORY_DIR
603 if not history_root.exists():
604 print(f"No history for: {loop_name}")
605 return 0
607 suffix = f"-{loop_name}"
608 all_entries: list[dict[str, Any]] = []
610 for run_dir in sorted(history_root.iterdir(), key=lambda d: d.name):
611 if not run_dir.is_dir() or not run_dir.name.endswith(suffix):
612 continue
613 meta_eval_file = run_dir / "meta-eval.jsonl"
614 if not meta_eval_file.exists():
615 continue
616 for line in meta_eval_file.read_text(encoding="utf-8").splitlines():
617 line = line.strip()
618 if line:
619 try:
620 all_entries.append(_json.loads(line))
621 except _json.JSONDecodeError:
622 pass
624 if not all_entries:
625 print(f"No meta-eval data for: {loop_name}")
626 return 0
628 total = len(all_entries)
629 agreed_count = sum(1 for e in all_entries if e.get("agreed") is True)
630 agreement_rate = agreed_count / total if total else 0.0
632 # Mean diff size per verdict
633 agreed_sizes = [
634 e.get("diff_stats", {}).get("files_changed", 0) or 0
635 for e in all_entries
636 if e.get("agreed") is True
637 ]
638 disagreed_sizes = [
639 e.get("diff_stats", {}).get("files_changed", 0) or 0
640 for e in all_entries
641 if e.get("agreed") is False
642 ]
643 mean_agreed = sum(agreed_sizes) / len(agreed_sizes) if agreed_sizes else 0.0
644 mean_disagreed = sum(disagreed_sizes) / len(disagreed_sizes) if disagreed_sizes else 0.0
646 # Detect streaks
647 flags: list[str] = []
648 optimistic_streak = 0
649 trivial_streak = 0
650 max_optimistic = 0
651 max_trivial = 0
653 for entry in all_entries:
654 agreed = entry.get("agreed")
655 files_changed = (entry.get("diff_stats") or {}).get("files_changed", 0) or 0
657 if agreed is False:
658 optimistic_streak += 1
659 trivial_streak = 0
660 elif agreed is True and files_changed == 0:
661 trivial_streak += 1
662 optimistic_streak = 0
663 else:
664 optimistic_streak = 0
665 trivial_streak = 0
667 max_optimistic = max(max_optimistic, optimistic_streak)
668 max_trivial = max(max_trivial, trivial_streak)
670 if max_optimistic >= 3:
671 flags.append(f"LLM optimistic drift detected (streak={max_optimistic})")
672 if max_trivial >= 3:
673 flags.append(f"Trivial agreement detected (streak={max_trivial})")
675 as_json = getattr(args, "json", False)
676 if as_json:
677 result = {
678 "loop": loop_name,
679 "total_entries": total,
680 "agreed_count": agreed_count,
681 "agreement_rate": round(agreement_rate, 4),
682 "mean_files_changed_when_agreed": round(mean_agreed, 2),
683 "mean_files_changed_when_disagreed": round(mean_disagreed, 2),
684 "max_optimistic_streak": max_optimistic,
685 "max_trivial_streak": max_trivial,
686 "flags": flags,
687 }
688 print_json(result)
689 else:
690 print(f"Meta-eval audit: {loop_name}")
691 print(f" Total entries: {total}")
692 print(f" Agreed: {agreed_count}/{total} ({agreement_rate:.0%})")
693 print(f" Mean Δfiles agreed: {mean_agreed:.1f}")
694 print(f" Mean Δfiles disagreed: {mean_disagreed:.1f}")
695 if flags:
696 print()
697 for flag in flags:
698 print(f" ⚠ {flag}")
699 else:
700 print(" No divergence flags.")
702 if as_json:
703 return 0 # let caller inspect JSON for flags; non-zero kills provider.runCli
704 return 1 if flags else 0
707def cmd_diagnose_evaluators(loop_name: str, args: argparse.Namespace, loops_dir: Path) -> int:
708 """Detect non-discriminating evaluators from run history.
710 Walks .loops/.history/*-{loop_name}/events.jsonl, computes per-state
711 Bernoulli variance p*(1-p) on verdicts, flags states below threshold
712 with pattern-matched recommendations.
714 Returns 0 if no states flagged, 1 if any low-variance evaluators found.
715 """
716 from little_loops.analytics.variance import compute_evaluator_variance
718 threshold = getattr(args, "threshold", 0.05)
719 min_runs = getattr(args, "min_runs", 10)
720 as_json = getattr(args, "json", False)
722 report = compute_evaluator_variance(loop_name, loops_dir, threshold, min_runs)
724 if report is None:
725 if not (loops_dir / ".history").exists():
726 print(f"No history for: {loop_name}")
727 elif len(list((loops_dir / ".history").glob(f"*-{loop_name}"))) < min_runs:
728 runs_found = len(list((loops_dir / ".history").glob(f"*-{loop_name}")))
729 print(
730 f"Insufficient history for: {loop_name} "
731 f"(found {runs_found} run(s), need {min_runs})"
732 )
733 else:
734 print(f"No evaluate events for: {loop_name}")
735 return 0
737 flagged = [s for s in report.states if s.variance < threshold]
739 if as_json:
740 print_json(report.to_dict())
741 else:
742 print(f"Evaluator Variance Report (n={report.total_runs} runs)")
743 if not report.states:
744 print(" No evaluator states found in run history.")
745 for state in report.states:
746 disc_label = "✓ discriminating" if state.variance >= threshold else "⚠ low variance"
747 print(
748 f" {state.state:20s} pass_rate={state.pass_rate:.2f} "
749 f"variance={state.variance:.2f} {disc_label}"
750 )
751 if state.recommendation:
752 for line in state.recommendation.split("\n"):
753 print(f" {line.strip()}")
755 if as_json:
756 return 0
757 return 1 if flagged else 0
760def cmd_promote_baseline(loop_name: str, args: argparse.Namespace, loops_dir: Path) -> int:
761 """Promote the latest run's action output as the new comparator baseline.
763 Reads action_output events from the most recent .history entry and writes
764 the concatenated output to .loops/baselines/<loop>/output.txt.
765 """
766 import json as _json
768 history_base = loops_dir / ".history"
769 if not history_base.exists():
770 print(f"No history for: {loop_name}")
771 return 1
773 suffix = f"-{loop_name}"
774 matched = sorted(
775 [d for d in history_base.iterdir() if d.is_dir() and d.name.endswith(suffix)],
776 key=lambda d: d.name,
777 reverse=True,
778 )
779 if not matched:
780 print(f"No history for: {loop_name}")
781 return 1
783 latest = matched[0]
784 events_file = latest / "events.jsonl"
785 if not events_file.exists():
786 print(f"No events.jsonl in latest run: {latest.name}")
787 return 1
789 lines = []
790 with open(events_file) as f:
791 for raw in f:
792 raw = raw.strip()
793 if not raw:
794 continue
795 try:
796 event = _json.loads(raw)
797 except _json.JSONDecodeError:
798 continue
799 if event.get("type") == "action_output":
800 line = event.get("line", "")
801 if line:
802 lines.append(line)
804 if not lines:
805 print(f"No action_output events in latest run: {latest.name}")
806 return 1
808 target = Path(loops_dir) / "baselines" / loop_name / "output.txt"
809 target.parent.mkdir(parents=True, exist_ok=True)
810 target.write_text("\n".join(lines))
811 print(f"Promoted baseline for {loop_name}: {target}")
812 return 0
815# ---------------------------------------------------------------------------
816# FSM diagram renderer — delegated to layout module (re-exported above)
817# ---------------------------------------------------------------------------
820# ---------------------------------------------------------------------------
821# State overview table
822# ---------------------------------------------------------------------------
825def _compact_transitions(state: StateConfig) -> str:
826 """Return a compact transition string for the overview table."""
827 raw: list[tuple[str, str]] = []
828 for label, target in [
829 ("yes", state.on_yes),
830 ("no", state.on_no),
831 ("error", state.on_error),
832 ("partial", state.on_partial),
833 ("next", state.next),
834 ]:
835 if target:
836 raw.append((label, target))
837 if state.route:
838 for verdict, target in state.route.routes.items():
839 raw.append((verdict, target))
840 if state.route.default:
841 raw.append(("_", state.route.default))
842 if not raw:
843 return "\u2014"
844 # Group by target, preserving first-seen order
845 seen: list[str] = []
846 by_target: dict[str, list[str]] = {}
847 for label, target in raw:
848 if target not in by_target:
849 seen.append(target)
850 by_target[target] = []
851 by_target[target].append(label)
852 return ", ".join(f"{'/'.join(by_target[t])}\u2192{t}" for t in seen)
855def _print_state_overview_table(fsm: FSMLoop) -> None:
856 """Print a compact summary table of all states."""
857 rows: list[tuple[str, str, str, str]] = []
858 for name, state in fsm.states.items():
859 # State name column
860 state_col = f"\u2192 {name}" if name == fsm.initial else f" {name}"
862 # Type column
863 if state.terminal:
864 type_col = "\u2014"
865 elif state.action_type:
866 type_col = state.action_type
867 elif state.action:
868 type_col = "shell"
869 else:
870 type_col = "\u2014"
872 # Action preview column
873 if state.terminal:
874 action_col = "(terminal)"
875 elif state.action:
876 src_lines = [ln.rstrip() for ln in state.action.strip().splitlines() if ln.rstrip()]
877 action_col = src_lines[0] if src_lines else "\u2014"
878 else:
879 action_col = "\u2014"
881 # Transitions column
882 trans_col = _compact_transitions(state)
883 rows.append((state_col, type_col, action_col, trans_col))
885 if not rows:
886 return
888 tw = terminal_width()
889 headers = ("State", "Type", "Action Preview", "Transitions")
890 col0_w = max(len(headers[0]), max(len(r[0]) for r in rows))
891 col1_w = max(len(headers[1]), max(len(r[1]) for r in rows))
892 # Remaining width split between action preview and transitions
893 fixed = col0_w + col1_w + 10 # margins + separators
894 remaining = max(20, tw - fixed)
895 col2_w = min(50, max(len(headers[2]), max(len(r[2]) for r in rows)), remaining * 3 // 5)
896 col2_w = max(10, col2_w)
897 col3_w = max(10, remaining - col2_w)
899 print(f" {headers[0]:<{col0_w}} {headers[1]:<{col1_w}} {headers[2]:<{col2_w}} {headers[3]}")
900 dash = "\u2500"
901 print(f" {dash * col0_w} {dash * col1_w} {dash * col2_w} {dash * col3_w}")
902 for state_col, type_col, action_col, trans_col in rows:
903 if len(action_col) > col2_w:
904 action_col = action_col[: col2_w - 1] + "\u2026"
905 if len(trans_col) > col3_w:
906 trans_col = trans_col[: col3_w - 1] + "\u2026"
907 colored_type = colorize(type_col, "2") if type_col == "\u2014" else type_col
908 print(
909 f" {state_col:<{col0_w}} {colored_type:<{col1_w}} "
910 f"{action_col:<{col2_w}} {trans_col}"
911 )
914# ---------------------------------------------------------------------------
915# cmd_show
916# ---------------------------------------------------------------------------
918_EVALUATE_TYPE_DISPLAY: dict[str, str] = {
919 "llm": "LLM",
920 "llm_structured": "LLM (structured)",
921 "exit_code": "exit code",
922 "output_numeric": "numeric",
923 "output_contains": "contains",
924 "output_json": "JSON",
925 "convergence": "convergence",
926 "diff_stall": "diff stall",
927 "action_stall": "action stall",
928 "comparator": "blind comparator",
929}
932def _humanize_evaluate_type(ev_type: str) -> str:
933 return _EVALUATE_TYPE_DISPLAY.get(ev_type, ev_type)
936def cmd_show(
937 loop_name: str,
938 args: argparse.Namespace,
939 loops_dir: Path,
940 logger: Logger,
941) -> int:
942 """Show loop details and structure."""
943 try:
944 path = resolve_loop_path(loop_name, loops_dir)
945 fsm, spec = load_loop_with_spec(loop_name, loops_dir, logger)
946 except FileNotFoundError as e:
947 logger.error(str(e))
948 return 1
949 except ValueError as e:
950 logger.error(f"Invalid loop: {e}")
951 return 1
953 if getattr(args, "json", False) and getattr(args, "show_diagrams", None) is not None:
954 logger.error("--json and --show-diagrams are mutually exclusive")
955 return 1
957 if getattr(args, "json", False):
958 data = fsm.to_dict()
959 if getattr(args, "resolved", False):
960 for state in data.get("states", {}).values():
961 if "loop" in state:
962 try:
963 child_path = resolve_loop_path(state["loop"], loops_dir)
964 child_fsm, _ = load_and_validate(child_path)
965 state["_subloop"] = child_fsm.to_dict().get("states", {})
966 except (FileNotFoundError, ValueError):
967 pass
968 print_json(data)
969 return 0
971 tw = terminal_width()
973 # Compute stats for header
974 n_states = len(fsm.states)
975 n_transitions = sum(
976 bool(s.on_yes)
977 + bool(s.on_no)
978 + bool(s.on_error)
979 + bool(s.on_partial)
980 + bool(s.next)
981 + bool(s.on_maintain)
982 + (len(s.route.routes) + bool(s.route.default) if s.route else 0)
983 for s in fsm.states.values()
984 )
986 # --- Compact metadata header ---
987 # Line 1: ── name ───────── N states · M transitions ──
988 stats_parts: list[str] = []
989 stats_parts.append(f"{n_states} states")
990 stats_parts.append(f"{n_transitions} transitions")
991 stats_str = " \u00b7 ".join(stats_parts)
993 header_left = f"\u2500\u2500 {loop_name} "
994 header_right = f" {stats_str} \u2500\u2500"
995 dashes = "\u2500" * max(0, tw - len(header_left) - len(header_right))
996 print(f"{header_left}{dashes}{header_right}")
998 # Line 2: source · max: N iter · handoff: X [· optional fields]
999 config_parts: list[str] = [str(path), f"max: {fsm.max_iterations} iter"]
1000 config_parts.append(f"handoff: {fsm.on_handoff}")
1001 if fsm.on_max_iterations is not None:
1002 config_parts.append(f"on_max_iterations: {fsm.on_max_iterations}")
1003 if fsm.timeout:
1004 config_parts.append(f"timeout: {fsm.timeout}s")
1005 if fsm.backoff:
1006 config_parts.append(f"backoff: {fsm.backoff}s")
1007 if fsm.maintain:
1008 config_parts.append("maintain: yes")
1009 if fsm.context:
1010 config_parts.append(f"context: {', '.join(fsm.context.keys())}")
1011 if fsm.scope:
1012 config_parts.append(f"scope: {', '.join(fsm.scope)}")
1013 llm = fsm.llm
1014 llm_parts = []
1015 if llm.model != "sonnet":
1016 llm_parts.append(f"model={llm.model}")
1017 if llm.max_tokens != 256:
1018 llm_parts.append(f"max_tokens={llm.max_tokens}")
1019 if llm.timeout != 30:
1020 llm_parts.append(f"timeout={llm.timeout}s")
1021 if llm_parts:
1022 config_parts.append(f"llm: {', '.join(llm_parts)}")
1023 if fsm.config is not None:
1024 cfg_parts = []
1025 if fsm.config.handoff_threshold is not None:
1026 cfg_parts.append(f"handoff_threshold={fsm.config.handoff_threshold}")
1027 if fsm.config.readiness_threshold is not None:
1028 cfg_parts.append(f"readiness_threshold={fsm.config.readiness_threshold}")
1029 if fsm.config.outcome_threshold is not None:
1030 cfg_parts.append(f"outcome_threshold={fsm.config.outcome_threshold}")
1031 if fsm.config.max_continuations is not None:
1032 cfg_parts.append(f"max_continuations={fsm.config.max_continuations}")
1033 if cfg_parts:
1034 config_parts.append(f"config: {', '.join(cfg_parts)}")
1035 imports = spec.get("import", [])
1036 if imports:
1037 config_parts.append(f"imports: {', '.join(imports)}")
1038 print(" " + " \u00b7 ".join(config_parts))
1040 # --- Description ---
1041 description = spec.get("description", "").strip()
1042 if description:
1043 print()
1044 print("Description:")
1045 for line in description.splitlines():
1046 print(f" {line}")
1048 # --- ASCII FSM Diagram ---
1049 verbose = getattr(args, "verbose", False)
1050 from pathlib import Path
1052 from little_loops.config import BRConfig
1054 badges = BRConfig(Path.cwd()).loops.glyphs.to_dict()
1055 facets = resolve_facets(args)
1056 print()
1057 print("Diagram:")
1058 if facets is None:
1059 diagram = _render_fsm_diagram(fsm, verbose=verbose, badges=badges)
1060 else:
1061 diagram = _render_fsm_diagram(
1062 fsm,
1063 badges=badges,
1064 mode=facets.scope,
1065 suppress_labels=not facets.edge_labels,
1066 title_only=facets.state_detail == "title",
1067 )
1068 if diagram:
1069 print(diagram)
1071 # --- State overview table ---
1072 print()
1073 _print_state_overview_table(fsm)
1075 # --- States & Transitions (verbose only) ---
1076 if verbose:
1077 print()
1078 print("States:")
1079 first_state = True
1080 for name, state in fsm.states.items():
1081 if not first_state:
1082 print()
1083 first_state = False
1085 # Improved state section header: ── name ──── MARKERS · type ──
1086 right_parts = []
1087 if name == fsm.initial:
1088 right_parts.append("INITIAL")
1089 if state.terminal:
1090 right_parts.append("TERMINAL")
1091 if state.action_type:
1092 right_parts.append(state.action_type)
1093 right_info = " \u00b7 ".join(right_parts)
1094 inner_left = f"\u2500\u2500 {name} "
1095 inner_right = f" {right_info} \u2500\u2500" if right_info else " \u2500\u2500"
1096 fill = "\u2500" * max(0, tw - 2 - len(inner_left) - len(inner_right))
1097 print(f" {inner_left}{fill}{inner_right}")
1099 if state.action:
1100 if verbose:
1101 indented = "\n ".join(state.action.strip().splitlines())
1102 print(f" action:\n {indented}")
1103 elif state.action_type == "prompt":
1104 lines_act = state.action.strip().splitlines()
1105 preview = "\n ".join(lines_act[:3])
1106 if len(lines_act) > 3 or len(state.action) > 200:
1107 preview += " ..."
1108 print(f" action:\n {preview}")
1109 else: # shell, slash_command, or None
1110 action_display = (
1111 state.action[:70] + "..." if len(state.action) > 70 else state.action
1112 )
1113 print(f" action: {action_display}")
1114 if state.evaluate:
1115 ev = state.evaluate
1116 print(f" evaluate: {_humanize_evaluate_type(ev.type)}")
1117 if ev.prompt:
1118 if verbose:
1119 print(" prompt:")
1120 for line in ev.prompt.strip().splitlines():
1121 print(f" \u2502 {line}")
1122 else:
1123 ev_lines = ev.prompt.strip().splitlines()
1124 preview = ev_lines[0][:100] + (
1125 " ..." if len(ev_lines) > 1 or len(ev_lines[0]) > 100 else ""
1126 )
1127 print(f" prompt: {preview}")
1128 if ev.min_confidence != 0.5:
1129 print(f" min_confidence: {ev.min_confidence}")
1130 if ev.operator:
1131 print(f" operator: {ev.operator} {ev.target}")
1132 if ev.pattern:
1133 print(f" pattern: {ev.pattern}")
1134 if state.capture:
1135 print(f" capture: {state.capture}")
1136 if state.timeout:
1137 print(f" timeout: {state.timeout}s")
1138 # Collect (label, target) pairs
1139 raw_transitions: list[tuple[str, str]] = []
1140 for label, target in [
1141 ("yes", state.on_yes),
1142 ("no", state.on_no),
1143 ("error", state.on_error),
1144 ("partial", state.on_partial),
1145 ("next", state.next),
1146 ("maintain", state.on_maintain),
1147 ]:
1148 if target:
1149 raw_transitions.append((label, target))
1150 if state.route:
1151 for verdict, target in state.route.routes.items():
1152 raw_transitions.append((verdict, target))
1153 if state.route.default:
1154 raw_transitions.append(("_", state.route.default))
1155 # Group by target, preserving first-seen order
1156 target_labels: dict[str, list[str]] = {}
1157 seen_targets: list[str] = []
1158 for label, target in raw_transitions:
1159 if target not in target_labels:
1160 target_labels[target] = []
1161 seen_targets.append(target)
1162 target_labels[target].append(label)
1163 transitions = [
1164 f"{_colorize_label('/'.join(target_labels[t]))} \u2500\u2500\u2192 {t}"
1165 for t in seen_targets
1166 ]
1167 if transitions:
1168 print(" Transitions:")
1169 for t in transitions:
1170 print(f" {t}")
1172 # --- Commands ---
1173 print()
1174 print("Commands:")
1175 if fsm.commands:
1176 cmds = [(e.cmd, e.comment) for e in fsm.commands]
1177 else:
1178 cmds = [
1179 (f"ll-loop run {loop_name}", "run"),
1180 (f"ll-loop test {loop_name}", "single test iteration"),
1181 (f"ll-loop stop {loop_name}", "stop a running loop"),
1182 (f"ll-loop status {loop_name}", "check if running"),
1183 (f"ll-loop history {loop_name}", "execution history"),
1184 ]
1185 col_width = max(len(c) for c, _ in cmds) + 2
1186 for cmd, comment in cmds:
1187 print(f" {cmd:<{col_width}} # {comment}")
1189 return 0
1192def cmd_fragments(
1193 lib_path: str,
1194 args: argparse.Namespace,
1195 loops_dir: Path,
1196 logger: Logger,
1197) -> int:
1198 """List fragments in a library file with their descriptions.
1200 Resolves the library path relative to loops_dir or the built-in loops directory,
1201 then prints a table of fragment names and their description fields.
1202 """
1203 import yaml
1205 # Resolve path: absolute/direct → loops_dir-relative → builtin
1206 p = Path(lib_path)
1207 if not p.exists():
1208 candidate = loops_dir / lib_path
1209 if candidate.exists():
1210 p = candidate
1211 else:
1212 builtin_candidate = get_builtin_loops_dir() / lib_path
1213 if builtin_candidate.exists():
1214 p = builtin_candidate
1215 else:
1216 logger.error(f"Fragment library not found: {lib_path}")
1217 return 1
1219 try:
1220 with open(p) as f:
1221 data = yaml.safe_load(f) or {}
1222 except Exception as e:
1223 logger.error(f"Failed to load library: {e}")
1224 return 1
1226 fragments: dict[str, Any] = data.get("fragments", {})
1227 if not fragments:
1228 print(f"No fragments defined in: {p}")
1229 return 0
1231 tw = terminal_width()
1232 print(colorize(f"Fragments in {p.name} ({len(fragments)}):", "1"))
1233 print()
1235 max_name_len = max(len(name) for name in fragments)
1236 name_col_w = max(max_name_len, 8)
1237 desc_col_w = max(20, tw - name_col_w - 6)
1239 header_name = "Fragment".ljust(name_col_w)
1240 print(f" {colorize(header_name, '1')} Description")
1241 print(f" {'─' * name_col_w} {'─' * desc_col_w}")
1243 for name, frag in fragments.items():
1244 raw_desc = frag.get("description", "") if isinstance(frag, dict) else ""
1245 raw_desc = raw_desc.strip()
1246 first_line = raw_desc.splitlines()[0] if raw_desc else ""
1247 if first_line and len(first_line) > desc_col_w:
1248 first_line = first_line[: desc_col_w - 1] + "\u2026"
1249 desc_str = first_line if first_line else colorize("(no description)", "2")
1250 padded_name = name.ljust(name_col_w)
1251 print(f" {colorize(padded_name, '36;1')} {desc_str}")
1253 return 0