Coverage for little_loops / cli / loop / info.py: 0%

1"""ll-loop info subcommands: list, history, show.""" 

2 

3from __future__ import annotations 

4 

5import argparse 

6import os 

7from datetime import datetime 

8from pathlib import Path 

9from typing import Any 

10 

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.layout import ( # noqa: F401 

17 _EDGE_LABEL_COLORS, 

18 _box_inner_lines, 

19 _colorize_diagram_labels, 

20 _colorize_label, 

21 _render_fsm_diagram, 

22) 

23from little_loops.cli.output import colorize, print_json, terminal_width 

24from little_loops.fsm.schema import FSMLoop, StateConfig 

25from little_loops.fsm.validation import load_and_validate 

26from little_loops.logger import Logger 

27 

28 

29def _load_loop_meta(path: Path) -> dict[str, Any]: 

30 """Return metadata from a loop YAML file (description, category, labels).""" 

31 import yaml 

32 

33 try: 

34 with open(path) as f: 

35 spec = yaml.safe_load(f) or {} 

36 desc_raw = spec.get("description", "") or "" 

37 desc = desc_raw.splitlines()[0] if desc_raw.strip() else "" 

38 category = spec.get("category", "") or "" 

39 labels: list[str] = spec.get("labels", []) or [] 

40 return {"description": desc, "category": category, "labels": labels} 

41 except Exception: 

42 return {"description": "", "category": "", "labels": []} 

43 

44 

45def cmd_list( 

46 args: argparse.Namespace, 

47 loops_dir: Path, 

48) -> int: 

49 """List loops.""" 

50 status_filter = getattr(args, "status", None) 

51 if getattr(args, "running", False) or status_filter: 

52 from little_loops.fsm.persistence import list_running_loops 

53 

54 states = list_running_loops(loops_dir) 

55 if status_filter: 

56 states = [s for s in states if s.status == status_filter] 

57 if not states: 

58 if status_filter: 

59 print(f"No loops with status: {status_filter}") 

60 return 1 

61 print("No running loops") 

62 return 0 

63 if getattr(args, "json", False): 

64 print_json([s.to_dict() for s in states]) 

65 return 0 

66 print(colorize("Running loops:", "1")) 

67 _STATUS_COLORS = {"running": "32", "interrupted": "33", "stopped": "2", "starting": "33"} 

68 

69 # Group by loop_name to avoid duplicate rows for multi-instance loops 

70 from collections import defaultdict 

71 

72 groups: dict[str, list] = defaultdict(list) 

73 for state in states: 

74 groups[state.loop_name].append(state) 

75 

76 for loop_name_key, group_states in groups.items(): 

77 if len(group_states) == 1: 

78 state = group_states[0] 

79 elapsed_s = (state.accumulated_ms or 0) // 1000 

80 elapsed_str = ( 

81 f"{elapsed_s}s" if elapsed_s < 60 else f"{elapsed_s // 60}m {elapsed_s % 60}s" 

82 ) 

83 name_str = colorize(state.loop_name, "1") 

84 state_str = colorize(state.current_state, "34") 

85 status_color = _STATUS_COLORS.get(state.status, "2") 

86 status_str = colorize(f"[{state.status}]", status_color) 

87 elapsed_colored = colorize(elapsed_str, "2") 

88 print( 

89 f" {name_str}: {state_str} (iteration {state.iteration})" 

90 f" {status_str} {elapsed_colored}" 

91 ) 

92 else: 

93 # Multiple instances: show a grouped summary 

94 name_str = colorize(loop_name_key, "1") 

95 statuses = ", ".join( 

96 colorize(f"[{s.status}]", _STATUS_COLORS.get(s.status, "2")) 

97 for s in group_states 

98 ) 

99 count_str = colorize(f"({len(group_states)} instances)", "2") 

100 print(f" {name_str}: {count_str} {statuses}") 

101 return 0 

102 

103 builtin_only = getattr(args, "builtin", False) 

104 

105 # Collect project loops (skipped when --builtin is set) 

106 project_names: set[str] = set() 

107 yaml_files: list[Path] = [] 

108 if not builtin_only and loops_dir.exists(): 

109 yaml_files = sorted(loops_dir.glob("*.yaml")) 

110 project_names = {p.stem for p in yaml_files} 

111 

112 # Collect built-in loops (excluding those overridden by project) 

113 builtin_dir = get_builtin_loops_dir() 

114 builtin_files: list[Path] = [] 

115 if builtin_dir.exists(): 

116 builtin_files = [ 

117 f for f in sorted(builtin_dir.glob("*.yaml")) if f.stem not in project_names 

118 ] 

119 

120 if not yaml_files and not builtin_files: 

121 if getattr(args, "json", False): 

122 print_json([]) 

123 return 0 

124 print("No loops available") 

125 return 0 

126 

127 # Build combined metadata list 

128 all_loops: list[dict[str, Any]] = [] 

129 for path in yaml_files: 

130 meta = _load_loop_meta(path) 

131 all_loops.append({"name": path.stem, "path": path, "builtin": False, **meta}) 

132 for path in builtin_files: 

133 meta = _load_loop_meta(path) 

134 all_loops.append({"name": path.stem, "path": path, "builtin": True, **meta}) 

135 

136 # Apply --category filter 

137 category_filter = getattr(args, "category", None) 

138 if category_filter: 

139 all_loops = [lp for lp in all_loops if lp["category"] == category_filter] 

140 

141 # Apply --label filter (action="append" → list or None) 

142 label_filters: list[str] = getattr(args, "label", None) or [] 

143 if label_filters: 

144 all_loops = [ 

145 lp 

146 for lp in all_loops 

147 if any(lf.lower() in [lb.lower() for lb in lp["labels"]] for lf in label_filters) 

148 ] 

149 

150 if not all_loops: 

151 if getattr(args, "json", False): 

152 print_json([]) 

153 return 0 

154 print("No loops match the given filters") 

155 return 0 

156 

157 if getattr(args, "json", False): 

158 items: list[dict[str, Any]] = [] 

159 for lp in all_loops: 

160 item: dict[str, Any] = { 

161 "name": lp["name"], 

162 "path": str(lp["path"]), 

163 "category": lp["category"], 

164 "labels": lp["labels"], 

165 } 

166 if lp["builtin"]: 

167 item["built_in"] = True 

168 items.append(item) 

169 print_json(items) 

170 return 0 

171 

172 # Human-readable: group by category 

173 buckets: dict[str, list[dict[str, Any]]] = {} 

174 for lp in all_loops: 

175 cat = lp["category"] or "uncategorized" 

176 if cat not in buckets: 

177 buckets[cat] = [] 

178 buckets[cat].append(lp) 

179 

180 # Sort categories; "uncategorized" always last 

181 sorted_cats = sorted(c for c in buckets if c != "uncategorized") 

182 if "uncategorized" in buckets: 

183 sorted_cats.append("uncategorized") 

184 

185 for cat in sorted_cats: 

186 group = buckets[cat] 

187 print(colorize(f"{cat} ({len(group)}):", "1")) 

188 for lp in group: 

189 name_str = colorize(lp["name"], "36;1") 

190 desc_str = f" {colorize(lp['description'], '2')}" if lp["description"] else "" 

191 tag_str = f" {colorize('[built-in]', '2')}" if lp["builtin"] else "" 

192 print(f" {name_str}{desc_str}{tag_str}") 

193 print() 

194 return 0 

195 

196 

197_EVENT_TYPE_WIDTH = 16 # width of "handoff_detected" 

198 

199 

200def _truncate(text: str, max_len: int) -> str: 

201 """Truncate text to max_len with ellipsis.""" 

202 if max_len < 1: 

203 return "" 

204 if len(text) <= max_len: 

205 return text 

206 return text[: max_len - 1] + "\u2026" 

207 

208 

209def _format_history_event( 

210 event: dict[str, Any], verbose: bool, width: int, full: bool = False 

211) -> str | None: 

212 """Format a single history event. Returns None to skip the event.""" 

213 raw_ts = event.get("ts", "") 

214 try: 

215 ts = datetime.fromisoformat(raw_ts).strftime("%H:%M:%S") 

216 except (ValueError, TypeError): 

217 ts = raw_ts[:8] if len(raw_ts) >= 8 else raw_ts.ljust(8) 

218 

219 event_type = event.get("event", "unknown") 

220 

221 if event_type == "action_output" and not verbose: 

222 return None 

223 

224 ts_str = colorize(ts, "2") 

225 etype_padded = event_type.ljust(_EVENT_TYPE_WIDTH) 

226 etype_color = "0" 

227 detail = "" 

228 extra_lines: list[str] = [] 

229 

230 # Indentation prefix for verbose sub-lines (aligns under event detail column) 

231 _indent = " " * (8 + 2 + _EVENT_TYPE_WIDTH + 2) 

232 

233 if event_type == "loop_start": 

234 etype_color = "1" 

235 detail = event.get("loop", "") 

236 

237 elif event_type == "loop_complete": 

238 etype_color = "1" 

239 final_state = event.get("final_state", "") 

240 iterations = event.get("iterations", "") 

241 terminated_by = event.get("terminated_by", "") 

242 detail = f"{final_state} {iterations} iter [{terminated_by}]" 

243 

244 elif event_type == "loop_resume": 

245 etype_color = "1" 

246 from_state = event.get("from_state", "") 

247 iteration = event.get("iteration", "") 

248 detail = f"from={from_state} iter={iteration}" 

249 

250 elif event_type == "state_enter": 

251 etype_color = "34" 

252 state = event.get("state", "") 

253 iteration = event.get("iteration", "") 

254 detail = f"{colorize(state, '1')} (iter {iteration})" 

255 

256 elif event_type == "action_start": 

257 action = event.get("action", "") 

258 is_prompt = event.get("is_prompt", False) 

259 kind_label = "prompt" if is_prompt else "shell" 

260 kind_str = colorize(f"[{kind_label}]", "2") 

261 first_line = ( 

262 next((ln.strip() for ln in action.splitlines() if ln.strip()), "") 

263 if is_prompt 

264 else action 

265 ) 

266 avail = width - 8 - 2 - _EVENT_TYPE_WIDTH - 2 - len(kind_label) - 2 - 2 

267 detail = f"{_truncate(first_line, max(avail, 20))} {kind_str}" 

268 

269 elif event_type == "action_output": 

270 # Only reached in verbose mode 

271 etype_color = "2" 

272 detail = colorize("\u2502 " + event.get("line", ""), "2") 

273 

274 elif event_type == "action_complete": 

275 exit_code = event.get("exit_code", 0) 

276 duration_ms = event.get("duration_ms", 0) 

277 if exit_code == 0: 

278 etype_color = "2" 

279 status_str = colorize("\u2713", "32") 

280 else: 

281 etype_color = "38;5;208" 

282 status_str = colorize(f"\u2717 exit={exit_code}", "38;5;208") 

283 detail = f"{status_str} {duration_ms}ms" 

284 is_prompt = event.get("is_prompt", False) 

285 session_jsonl = event.get("session_jsonl") if is_prompt else None 

286 if session_jsonl: 

287 session_display = session_jsonl if verbose else os.path.basename(session_jsonl) 

288 detail += f" session={colorize(session_display, '2')}" 

289 if verbose: 

290 output_preview = event.get("output_preview", "") 

291 if output_preview: 

292 avail_w = width - len(_indent) - 2 

293 preview_text = ( 

294 output_preview if full else _truncate(output_preview, max(avail_w, 40)) 

295 ) 

296 for preview_line in preview_text.splitlines()[:5]: 

297 extra_lines.append(colorize(_indent + "\u2502 " + preview_line, "2")) 

298 

299 elif event_type == "evaluate": 

300 verdict = event.get("verdict", "") 

301 confidence = event.get("confidence", "") 

302 reason = event.get("reason", "") 

303 if verdict == "yes": 

304 etype_color = "32" 

305 verdict_str = colorize("\u2713 yes", "32") 

306 else: 

307 etype_color = "38;5;208" 

308 verdict_str = colorize(f"\u2717 {verdict}", "38;5;208") 

309 conf_part = f" confidence={confidence}" if confidence != "" else "" 

310 avail = width - 8 - 2 - _EVENT_TYPE_WIDTH - 2 - len("\u2713 yes") - len(conf_part) - 2 

311 reason_part = f" {_truncate(reason, max(avail, 20))}" if reason else "" 

312 detail = f"{verdict_str}{conf_part}{reason_part}" 

313 if verbose: 

314 llm_model = event.get("llm_model", "") 

315 llm_latency_ms = event.get("llm_latency_ms", "") 

316 llm_prompt = event.get("llm_prompt", "") 

317 llm_raw_output = event.get("llm_raw_output", "") 

318 if llm_model or llm_prompt: 

319 meta_parts = [] 

320 if llm_model: 

321 meta_parts.append(f"model={llm_model}") 

322 if llm_latency_ms != "": 

323 meta_parts.append(f"latency={llm_latency_ms}ms") 

324 meta_str = " ".join(meta_parts) 

325 extra_lines.append( 

326 colorize(_indent + colorize("LLM Call", "2") + " " + meta_str, "2") 

327 ) 

328 avail_w = width - len(_indent) - len("Prompt: ") - 2 

329 if llm_prompt: 

330 prompt_text = llm_prompt if full else _truncate(llm_prompt, max(avail_w, 40)) 

331 extra_lines.append(colorize(_indent + "Prompt: " + prompt_text, "2")) 

332 if llm_raw_output: 

333 resp_text = ( 

334 llm_raw_output if full else _truncate(llm_raw_output, max(avail_w, 40)) 

335 ) 

336 extra_lines.append(colorize(_indent + "Response: " + resp_text, "2")) 

337 

338 elif event_type == "route": 

339 etype_color = "2" 

340 from_state = event.get("from", "") 

341 to_state = event.get("to", "") 

342 detail = f"{from_state} \u2192 {colorize(to_state, '34')}" 

343 

344 elif event_type == "handoff_detected": 

345 etype_color = "33" 

346 detail = f"state={event.get('state', '')} iter={event.get('iteration', '')}" 

347 

348 else: 

349 details = {k: v for k, v in event.items() if k not in ("event", "ts")} 

350 detail = " ".join(f"{k}={v}" for k, v in details.items()) 

351 

352 etype_str = colorize(etype_padded, etype_color) 

353 main_line = f"{ts_str} {etype_str} {detail}" 

354 if extra_lines: 

355 return "\n".join([main_line] + extra_lines) 

356 return main_line 

357 

358 

359def _format_duration(ms: int) -> str: 

360 """Format milliseconds as a human-readable duration.""" 

361 if ms < 1000: 

362 return f"{ms}ms" 

363 s = ms // 1000 

364 if s < 60: 

365 return f"{s}s" 

366 m, s = divmod(s, 60) 

367 if m < 60: 

368 return f"{m}m{s:02d}s" 

369 h, m = divmod(m, 60) 

370 return f"{h}h{m:02d}m{s:02d}s" 

371 

372 

373def _list_archived_runs(loop_name: str, loops_dir: Path, as_json: bool) -> int: 

374 """List archived runs for a loop.""" 

375 import json as _json 

376 

377 from little_loops.fsm.persistence import HISTORY_DIR, LoopState 

378 

379 history_base = loops_dir / HISTORY_DIR 

380 if not history_base.exists(): 

381 print(f"No history for: {loop_name}") 

382 return 0 

383 

384 # Flat layout: run dirs are <run_id>-<loop_name> directly under .history/ 

385 suffix = f"-{loop_name}" 

386 runs: list[tuple[str, LoopState | None]] = [] 

387 for run_dir in sorted(history_base.iterdir(), key=lambda d: d.name, reverse=True): 

388 if not run_dir.is_dir() or not run_dir.name.endswith(suffix): 

389 continue 

390 run_id = run_dir.name[: -len(suffix)] 

391 state_file = run_dir / "state.json" 

392 state: LoopState | None = None 

393 if state_file.exists(): 

394 try: 

395 data = _json.loads(state_file.read_text()) 

396 state = LoopState.from_dict(data) 

397 except (ValueError, KeyError): 

398 pass 

399 runs.append((run_id, state)) 

400 

401 if not runs: 

402 print(f"No history for: {loop_name}") 

403 return 0 

404 

405 if as_json: 

406 print( 

407 _json.dumps( 

408 [ 

409 { 

410 "run_id": rid, 

411 "status": s.status if s else None, 

412 "started_at": s.started_at if s else None, 

413 "iterations": s.iteration if s else None, 

414 "duration_ms": s.accumulated_ms if s else None, 

415 } 

416 for rid, s in runs 

417 ], 

418 indent=2, 

419 ) 

420 ) 

421 return 0 

422 

423 status_colors = { 

424 "completed": "\033[32m", 

425 "failed": "\033[31m", 

426 "interrupted": "\033[33m", 

427 "awaiting_continuation": "\033[36m", 

428 "timed_out": "\033[33m", 

429 "running": "\033[34m", 

430 } 

431 reset = "\033[0m" 

432 

433 print(f"Archived runs for: {loop_name} ({len(runs)} total)") 

434 print() 

435 

436 for run_id, state in runs: 

437 if state is not None: 

438 color = status_colors.get(state.status, "") 

439 status_str = f"{color}{state.status}{reset}" 

440 duration_str = _format_duration(state.accumulated_ms) if state.accumulated_ms else "?" 

441 started = state.started_at[:19].replace("T", " ") if state.started_at else "?" 

442 iters = f"{state.iteration} iters" 

443 else: 

444 status_str = "unknown" 

445 duration_str = "?" 

446 started = "?" 

447 iters = "?" 

448 print(f" {run_id} {status_str} {started} {iters} {duration_str}") 

449 

450 print() 

451 print(f"To view events: ll-loop history {loop_name} <run-id>") 

452 return 0 

453 

454 

455def cmd_history( 

456 loop_name: str, 

457 run_id: str | None, 

458 args: argparse.Namespace, 

459 loops_dir: Path, 

460) -> int: 

461 """Show loop history. 

462 

463 Without run_id: lists all archived runs with status and duration. 

464 With run_id: shows events for that specific archived run. 

465 """ 

466 tail = getattr(args, "tail", 50) 

467 full = getattr(args, "full", False) 

468 verbose = getattr(args, "verbose", False) or full 

469 as_json = getattr(args, "json", False) 

470 

471 if run_id is None: 

472 return _list_archived_runs(loop_name, loops_dir, as_json) 

473 

474 # Show events for a specific archived run 

475 from little_loops.fsm.persistence import get_archived_events 

476 

477 events = get_archived_events(loop_name, run_id, loops_dir) 

478 

479 if not events: 

480 print(f"No events found for run {run_id} of loop {loop_name}") 

481 return 1 

482 

483 w = terminal_width() 

484 if not verbose: 

485 events = [e for e in events if e.get("event") != "action_output"] 

486 

487 # Apply optional filters (before --tail slice) 

488 event_filter = getattr(args, "event", None) 

489 if event_filter: 

490 events = [e for e in events if e.get("event") == event_filter] 

491 

492 state_filter = getattr(args, "state", None) 

493 if state_filter: 

494 events = [ 

495 e 

496 for e in events 

497 if ( 

498 e.get("state") == state_filter 

499 or e.get("from") == state_filter 

500 or e.get("to") == state_filter 

501 ) 

502 ] 

503 

504 since_str = getattr(args, "since", None) 

505 if since_str: 

506 from datetime import timedelta 

507 

508 from little_loops.text_utils import parse_duration 

509 

510 cutoff = datetime.now() - timedelta(seconds=parse_duration(since_str)) 

511 events = [ 

512 e 

513 for e in events 

514 if datetime.fromisoformat(e["ts"].replace("Z", "+00:00")).replace(tzinfo=None) >= cutoff 

515 ] 

516 

517 if as_json: 

518 print_json(events[-tail:]) 

519 return 0 

520 for event in events[-tail:]: 

521 line = _format_history_event(event, verbose, w, full=full) 

522 if line is not None: 

523 print(line) 

524 

525 return 0 

526 

527 

528# --------------------------------------------------------------------------- 

529# FSM diagram renderer — delegated to layout module (re-exported above) 

530# --------------------------------------------------------------------------- 

531 

532 

533# --------------------------------------------------------------------------- 

534# State overview table 

535# --------------------------------------------------------------------------- 

536 

537 

538def _compact_transitions(state: StateConfig) -> str: 

539 """Return a compact transition string for the overview table.""" 

540 raw: list[tuple[str, str]] = [] 

541 for label, target in [ 

542 ("yes", state.on_yes), 

543 ("no", state.on_no), 

544 ("error", state.on_error), 

545 ("partial", state.on_partial), 

546 ("next", state.next), 

547 ]: 

548 if target: 

549 raw.append((label, target)) 

550 if state.route: 

551 for verdict, target in state.route.routes.items(): 

552 raw.append((verdict, target)) 

553 if state.route.default: 

554 raw.append(("_", state.route.default)) 

555 if not raw: 

556 return "\u2014" 

557 # Group by target, preserving first-seen order 

558 seen: list[str] = [] 

559 by_target: dict[str, list[str]] = {} 

560 for label, target in raw: 

561 if target not in by_target: 

562 seen.append(target) 

563 by_target[target] = [] 

564 by_target[target].append(label) 

565 return ", ".join(f"{'/'.join(by_target[t])}\u2192{t}" for t in seen) 

566 

567 

568def _print_state_overview_table(fsm: FSMLoop) -> None: 

569 """Print a compact summary table of all states.""" 

570 rows: list[tuple[str, str, str, str]] = [] 

571 for name, state in fsm.states.items(): 

572 # State name column 

573 state_col = f"\u2192 {name}" if name == fsm.initial else f" {name}" 

574 

575 # Type column 

576 if state.terminal: 

577 type_col = "\u2014" 

578 elif state.action_type: 

579 type_col = state.action_type 

580 elif state.action: 

581 type_col = "shell" 

582 else: 

583 type_col = "\u2014" 

584 

585 # Action preview column 

586 if state.terminal: 

587 action_col = "(terminal)" 

588 elif state.action: 

589 src_lines = [ln.rstrip() for ln in state.action.strip().splitlines() if ln.rstrip()] 

590 action_col = src_lines[0] if src_lines else "\u2014" 

591 else: 

592 action_col = "\u2014" 

593 

594 # Transitions column 

595 trans_col = _compact_transitions(state) 

596 rows.append((state_col, type_col, action_col, trans_col)) 

597 

598 if not rows: 

599 return 

600 

601 tw = terminal_width() 

602 headers = ("State", "Type", "Action Preview", "Transitions") 

603 col0_w = max(len(headers[0]), max(len(r[0]) for r in rows)) 

604 col1_w = max(len(headers[1]), max(len(r[1]) for r in rows)) 

605 # Remaining width split between action preview and transitions 

606 fixed = col0_w + col1_w + 10 # margins + separators 

607 remaining = max(20, tw - fixed) 

608 col2_w = min(50, max(len(headers[2]), max(len(r[2]) for r in rows)), remaining * 3 // 5) 

609 col2_w = max(10, col2_w) 

610 col3_w = max(10, remaining - col2_w) 

611 

612 print(f" {headers[0]:<{col0_w}} {headers[1]:<{col1_w}} {headers[2]:<{col2_w}} {headers[3]}") 

613 dash = "\u2500" 

614 print(f" {dash * col0_w} {dash * col1_w} {dash * col2_w} {dash * col3_w}") 

615 for state_col, type_col, action_col, trans_col in rows: 

616 if len(action_col) > col2_w: 

617 action_col = action_col[: col2_w - 1] + "\u2026" 

618 if len(trans_col) > col3_w: 

619 trans_col = trans_col[: col3_w - 1] + "\u2026" 

620 colored_type = colorize(type_col, "2") if type_col == "\u2014" else type_col 

621 print( 

622 f" {state_col:<{col0_w}} {colored_type:<{col1_w}} " 

623 f"{action_col:<{col2_w}} {trans_col}" 

624 ) 

625 

626 

627# --------------------------------------------------------------------------- 

628# cmd_show 

629# --------------------------------------------------------------------------- 

630 

631_EVALUATE_TYPE_DISPLAY: dict[str, str] = { 

632 "llm": "LLM", 

633 "llm_structured": "LLM (structured)", 

634 "exit_code": "exit code", 

635 "output_numeric": "numeric", 

636 "output_contains": "contains", 

637 "output_json": "JSON", 

638 "convergence": "convergence", 

639} 

640 

641 

642def _humanize_evaluate_type(ev_type: str) -> str: 

643 return _EVALUATE_TYPE_DISPLAY.get(ev_type, ev_type) 

644 

645 

646def cmd_show( 

647 loop_name: str, 

648 args: argparse.Namespace, 

649 loops_dir: Path, 

650 logger: Logger, 

651) -> int: 

652 """Show loop details and structure.""" 

653 try: 

654 path = resolve_loop_path(loop_name, loops_dir) 

655 fsm, spec = load_loop_with_spec(loop_name, loops_dir, logger) 

656 except FileNotFoundError as e: 

657 logger.error(str(e)) 

658 return 1 

659 except ValueError as e: 

660 logger.error(f"Invalid loop: {e}") 

661 return 1 

662 

663 if getattr(args, "json", False): 

664 data = fsm.to_dict() 

665 if getattr(args, "resolved", False): 

666 for state in data.get("states", {}).values(): 

667 if "loop" in state: 

668 try: 

669 child_path = resolve_loop_path(state["loop"], loops_dir) 

670 child_fsm, _ = load_and_validate(child_path) 

671 state["_subloop"] = child_fsm.to_dict().get("states", {}) 

672 except (FileNotFoundError, ValueError): 

673 pass 

674 print_json(data) 

675 return 0 

676 

677 tw = terminal_width() 

678 

679 # Compute stats for header 

680 n_states = len(fsm.states) 

681 n_transitions = sum( 

682 bool(s.on_yes) 

683 + bool(s.on_no) 

684 + bool(s.on_error) 

685 + bool(s.on_partial) 

686 + bool(s.next) 

687 + bool(s.on_maintain) 

688 + (len(s.route.routes) + bool(s.route.default) if s.route else 0) 

689 for s in fsm.states.values() 

690 ) 

691 

692 # --- Compact metadata header --- 

693 # Line 1: ── name ───────── N states · M transitions ── 

694 stats_parts: list[str] = [] 

695 stats_parts.append(f"{n_states} states") 

696 stats_parts.append(f"{n_transitions} transitions")