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

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.logger import Logger 

26 

27 

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

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

30 import yaml 

31 

32 try: 

33 with open(path) as f: 

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

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

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

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

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

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

40 except Exception: 

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

42 

43 

44def cmd_list( 

45 args: argparse.Namespace, 

46 loops_dir: Path, 

47) -> int: 

48 """List loops.""" 

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

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

51 from little_loops.fsm.persistence import list_running_loops 

52 

53 states = list_running_loops(loops_dir) 

54 if status_filter: 

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

56 if not states: 

57 if status_filter: 

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

59 return 1 

60 print("No running loops") 

61 return 0 

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

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

64 return 0 

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

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

67 for state in states: 

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

69 elapsed_str = ( 

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

71 ) 

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

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

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

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

76 elapsed_colored = colorize(elapsed_str, "2") 

77 print( 

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

79 f" {status_str} {elapsed_colored}" 

80 ) 

81 return 0 

82 

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

84 

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

86 project_names: set[str] = set() 

87 yaml_files: list[Path] = [] 

88 if not builtin_only and loops_dir.exists(): 

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

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

91 

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

93 builtin_dir = get_builtin_loops_dir() 

94 builtin_files: list[Path] = [] 

95 if builtin_dir.exists(): 

96 builtin_files = [ 

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

98 ] 

99 

100 if not yaml_files and not builtin_files: 

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

102 print_json([]) 

103 return 0 

104 print("No loops available") 

105 return 0 

106 

107 # Build combined metadata list 

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

109 for path in yaml_files: 

110 meta = _load_loop_meta(path) 

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

112 for path in builtin_files: 

113 meta = _load_loop_meta(path) 

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

115 

116 # Apply --category filter 

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

118 if category_filter: 

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

120 

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

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

123 if label_filters: 

124 all_loops = [ 

125 lp 

126 for lp in all_loops 

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

128 ] 

129 

130 if not all_loops: 

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

132 print_json([]) 

133 return 0 

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

135 return 0 

136 

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

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

139 for lp in all_loops: 

140 item: dict[str, Any] = { 

141 "name": lp["name"], 

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

143 "category": lp["category"], 

144 "labels": lp["labels"], 

145 } 

146 if lp["builtin"]: 

147 item["built_in"] = True 

148 items.append(item) 

149 print_json(items) 

150 return 0 

151 

152 # Human-readable: group by category 

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

154 for lp in all_loops: 

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

156 if cat not in buckets: 

157 buckets[cat] = [] 

158 buckets[cat].append(lp) 

159 

160 # Sort categories; "uncategorized" always last 

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

162 if "uncategorized" in buckets: 

163 sorted_cats.append("uncategorized") 

164 

165 for cat in sorted_cats: 

166 group = buckets[cat] 

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

168 for lp in group: 

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

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

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

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

173 print() 

174 return 0 

175 

176 

177_EVENT_TYPE_WIDTH = 16 # width of "handoff_detected" 

178 

179 

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

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

182 if max_len < 1: 

183 return "" 

184 if len(text) <= max_len: 

185 return text 

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

187 

188 

189def _format_history_event( 

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

191) -> str | None: 

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

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

194 try: 

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

196 except (ValueError, TypeError): 

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

198 

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

200 

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

202 return None 

203 

204 ts_str = colorize(ts, "2") 

205 etype_padded = event_type.ljust(_EVENT_TYPE_WIDTH) 

206 etype_color = "0" 

207 detail = "" 

208 extra_lines: list[str] = [] 

209 

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

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

212 

213 if event_type == "loop_start": 

214 etype_color = "1" 

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

216 

217 elif event_type == "loop_complete": 

218 etype_color = "1" 

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

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

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

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

223 

224 elif event_type == "loop_resume": 

225 etype_color = "1" 

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

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

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

229 

230 elif event_type == "state_enter": 

231 etype_color = "34" 

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

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

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

235 

236 elif event_type == "action_start": 

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

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

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

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

241 first_line = ( 

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

243 if is_prompt 

244 else action 

245 ) 

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

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

248 

249 elif event_type == "action_output": 

250 # Only reached in verbose mode 

251 etype_color = "2" 

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

253 

254 elif event_type == "action_complete": 

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

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

257 if exit_code == 0: 

258 etype_color = "2" 

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

260 else: 

261 etype_color = "38;5;208" 

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

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

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

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

266 if session_jsonl: 

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

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

269 if verbose: 

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

271 if output_preview: 

272 avail_w = width - len(_indent) - 2 

273 preview_text = ( 

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

275 ) 

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

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

278 

279 elif event_type == "evaluate": 

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

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

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

283 if verdict == "yes": 

284 etype_color = "32" 

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

286 else: 

287 etype_color = "38;5;208" 

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

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

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

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

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

293 if verbose: 

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

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

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

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

298 if llm_model or llm_prompt: 

299 meta_parts = [] 

300 if llm_model: 

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

302 if llm_latency_ms != "": 

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

304 meta_str = " ".join(meta_parts) 

305 extra_lines.append( 

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

307 ) 

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

309 if llm_prompt: 

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

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

312 if llm_raw_output: 

313 resp_text = ( 

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

315 ) 

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

317 

318 elif event_type == "route": 

319 etype_color = "2" 

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

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

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

323 

324 elif event_type == "handoff_detected": 

325 etype_color = "33" 

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

327 

328 else: 

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

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

331 

332 etype_str = colorize(etype_padded, etype_color) 

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

334 if extra_lines: 

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

336 return main_line 

337 

338 

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

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

341 if ms < 1000: 

342 return f"{ms}ms" 

343 s = ms // 1000 

344 if s < 60: 

345 return f"{s}s" 

346 m, s = divmod(s, 60) 

347 if m < 60: 

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

349 h, m = divmod(m, 60) 

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

351 

352 

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

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

355 import json as _json 

356 

357 from little_loops.fsm.persistence import HISTORY_DIR, LoopState 

358 

359 history_base = loops_dir / HISTORY_DIR 

360 if not history_base.exists(): 

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

362 return 0 

363 

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

365 suffix = f"-{loop_name}" 

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

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

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

369 continue 

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

371 state_file = run_dir / "state.json" 

372 state: LoopState | None = None 

373 if state_file.exists(): 

374 try: 

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

376 state = LoopState.from_dict(data) 

377 except (ValueError, KeyError): 

378 pass 

379 runs.append((run_id, state)) 

380 

381 if not runs: 

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

383 return 0 

384 

385 if as_json: 

386 print( 

387 _json.dumps( 

388 [ 

389 { 

390 "run_id": rid, 

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

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

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

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

395 } 

396 for rid, s in runs 

397 ], 

398 indent=2, 

399 ) 

400 ) 

401 return 0 

402 

403 status_colors = { 

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

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

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

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

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

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

410 } 

411 reset = "\033[0m" 

412 

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

414 print() 

415 

416 for run_id, state in runs: 

417 if state is not None: 

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

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

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

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

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

423 else: 

424 status_str = "unknown" 

425 duration_str = "?" 

426 started = "?" 

427 iters = "?" 

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

429 

430 print() 

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

432 return 0 

433 

434 

435def cmd_history( 

436 loop_name: str, 

437 run_id: str | None, 

438 args: argparse.Namespace, 

439 loops_dir: Path, 

440) -> int: 

441 """Show loop history. 

442 

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

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

445 """ 

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

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

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

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

450 

451 if run_id is None: 

452 return _list_archived_runs(loop_name, loops_dir, as_json) 

453 

454 # Show events for a specific archived run 

455 from little_loops.fsm.persistence import get_archived_events 

456 

457 events = get_archived_events(loop_name, run_id, loops_dir) 

458 

459 if not events: 

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

461 return 1 

462 

463 w = terminal_width() 

464 if not verbose: 

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

466 

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

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

469 if event_filter: 

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

471 

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

473 if state_filter: 

474 events = [ 

475 e 

476 for e in events 

477 if ( 

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

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

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

481 ) 

482 ] 

483 

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

485 if since_str: 

486 from datetime import timedelta 

487 

488 from little_loops.text_utils import parse_duration 

489 

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

491 events = [ 

492 e 

493 for e in events 

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

495 ] 

496 

497 if as_json: 

498 print_json(events[-tail:]) 

499 return 0 

500 for event in events[-tail:]: 

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

502 if line is not None: 

503 print(line) 

504 

505 return 0 

506 

507 

508# --------------------------------------------------------------------------- 

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

510# --------------------------------------------------------------------------- 

511 

512 

513# --------------------------------------------------------------------------- 

514# State overview table 

515# --------------------------------------------------------------------------- 

516 

517 

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

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

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

521 for label, target in [ 

522 ("yes", state.on_yes), 

523 ("no", state.on_no), 

524 ("error", state.on_error), 

525 ("partial", state.on_partial), 

526 ("next", state.next), 

527 ]: 

528 if target: 

529 raw.append((label, target)) 

530 if state.route: 

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

532 raw.append((verdict, target)) 

533 if state.route.default: 

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

535 if not raw: 

536 return "\u2014" 

537 # Group by target, preserving first-seen order 

538 seen: list[str] = [] 

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

540 for label, target in raw: 

541 if target not in by_target: 

542 seen.append(target) 

543 by_target[target] = [] 

544 by_target[target].append(label) 

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

546 

547 

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

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

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

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

552 # State name column 

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

554 

555 # Type column 

556 if state.terminal: 

557 type_col = "\u2014" 

558 elif state.action_type: 

559 type_col = state.action_type 

560 elif state.action: 

561 type_col = "shell" 

562 else: 

563 type_col = "\u2014" 

564 

565 # Action preview column 

566 if state.terminal: 

567 action_col = "(terminal)" 

568 elif state.action: 

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

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

571 else: 

572 action_col = "\u2014" 

573 

574 # Transitions column 

575 trans_col = _compact_transitions(state) 

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

577 

578 if not rows: 

579 return 

580 

581 tw = terminal_width() 

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

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

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

585 # Remaining width split between action preview and transitions 

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

587 remaining = max(20, tw - fixed) 

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

589 col2_w = max(10, col2_w) 

590 col3_w = max(10, remaining - col2_w) 

591 

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

593 dash = "\u2500" 

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

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

596 if len(action_col) > col2_w: 

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

598 if len(trans_col) > col3_w: 

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

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

601 print( 

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

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

604 ) 

605 

606 

607# --------------------------------------------------------------------------- 

608# cmd_show 

609# --------------------------------------------------------------------------- 

610 

611_EVALUATE_TYPE_DISPLAY: dict[str, str] = { 

612 "llm": "LLM", 

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

614 "exit_code": "exit code", 

615 "output_numeric": "numeric", 

616 "output_contains": "contains", 

617 "output_json": "JSON", 

618 "convergence": "convergence", 

619} 

620 

621 

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

623 return _EVALUATE_TYPE_DISPLAY.get(ev_type, ev_type) 

624 

625 

626def cmd_show( 

627 loop_name: str, 

628 args: argparse.Namespace, 

629 loops_dir: Path, 

630 logger: Logger, 

631) -> int: 

632 """Show loop details and structure.""" 

633 try: 

634 path = resolve_loop_path(loop_name, loops_dir) 

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

636 except FileNotFoundError as e: 

637 logger.error(str(e)) 

638 return 1 

639 except ValueError as e: 

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

641 return 1 

642 

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

644 print_json(fsm.to_dict()) 

645 return 0 

646 

647 tw = terminal_width() 

648 

649 # Compute stats for header 

650 n_states = len(fsm.states) 

651 n_transitions = sum( 

652 bool(s.on_yes) 

653 + bool(s.on_no) 

654 + bool(s.on_error) 

655 + bool(s.on_partial) 

656 + bool(s.next) 

657 + bool(s.on_maintain) 

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

659 for s in fsm.states.values() 

660 ) 

661 

662 # --- Compact metadata header --- 

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

664 stats_parts: list[str] = [] 

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

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

667 stats_str = " \u00b7 ".join(stats_parts) 

668 

669 header_left = f"\u2500\u2500 {loop_name} " 

670 header_right = f" {stats_str} \u2500\u2500" 

671 dashes = "\u2500" * max(0, tw - len(header_left) - len(header_right)) 

672 print(f"{header_left}{dashes}{header_right}") 

673 

674 # Line 2: source · max: N iter · handoff: X [· optional fields] 

675 config_parts: list[str] = [str(path), f"max: {fsm.max_iterations} iter"] 

676 config_parts.append(f"handoff: {fsm.on_handoff}") 

677 if fsm.timeout: 

678 config_parts.append(f"timeout: {fsm.timeout}s") 

679 if fsm.backoff: 

680 config_parts.append(f"backoff: {fsm.backoff}s") 

681 if fsm.maintain: 

682 config_parts.append("maintain: yes") 

683 if fsm.context: 

684 config_parts.append(f"context: {', '.join(fsm.context.keys())}") 

685 if fsm.scope: 

686 config_parts.append(f"scope: {', '.join(fsm.scope)}") 

687 llm = fsm.llm 

688 llm_parts = [] 

689 if llm.model != "sonnet": 

690 llm_parts.append(f"model={llm.model}") 

691 if llm.max_tokens != 256: 

692 llm_parts.append(f"max_tokens={llm.max_tokens}")