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 

7import re 

8from datetime import datetime 

9from pathlib import Path 

10from typing import Any 

11 

12from little_loops.cli.loop._helpers import ( 

13 get_builtin_loops_dir, 

14 load_loop_with_spec, 

15 resolve_loop_path, 

16) 

17from little_loops.cli.loop.diagram_modes import resolve_facets 

18from little_loops.cli.loop.layout import ( # noqa: F401 

19 _EDGE_LABEL_COLORS, 

20 _box_inner_lines, 

21 _colorize_diagram_labels, 

22 _colorize_label, 

23 _render_fsm_diagram, 

24) 

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

26from little_loops.fsm import is_runnable_loop 

27from little_loops.fsm.schema import FSMLoop, StateConfig 

28from little_loops.fsm.validation import load_and_validate 

29from little_loops.logger import Logger 

30 

31 

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

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

34 import yaml 

35 

36 try: 

37 with open(path) as f: 

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

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

40 if desc_raw.strip(): 

41 raw_lines = desc_raw.splitlines() 

42 desc = raw_lines[0] 

43 if len(raw_lines) > 1: 

44 desc += "…" 

45 else: 

46 desc = "" 

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

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

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

50 except Exception: 

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

52 

53 

54def cmd_list( 

55 args: argparse.Namespace, 

56 loops_dir: Path, 

57) -> int: 

58 """List loops.""" 

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

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

61 from little_loops.fsm.persistence import list_running_loops 

62 

63 states = list_running_loops(loops_dir) 

64 if status_filter: 

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

66 if not states: 

67 if status_filter: 

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

69 return 1 

70 print("No running loops") 

71 return 0 

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

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

74 return 0 

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

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

77 

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

79 from collections import defaultdict 

80 

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

82 for state in states: 

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

84 

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

86 if len(group_states) == 1: 

87 state = group_states[0] 

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

89 elapsed_str = ( 

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

91 ) 

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

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

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

95 display_status = "paused" if state.status == "interrupted" else state.status 

96 status_str = colorize(f"[{display_status}]", status_color) 

97 elapsed_colored = colorize(elapsed_str, "2") 

98 print( 

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

100 f" {status_str} {elapsed_colored}" 

101 ) 

102 else: 

103 # Multiple instances: show a grouped summary 

104 name_str = colorize(loop_name_key, "1") 

105 statuses = ", ".join( 

106 colorize( 

107 f"[{'paused' if s.status == 'interrupted' else s.status}]", 

108 _STATUS_COLORS.get(s.status, "2"), 

109 ) 

110 for s in group_states 

111 ) 

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

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

114 return 0 

115 

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

117 

118 def _rel_key(path: Path, base: Path) -> str: 

119 """Relative-path identifier matching what `ll-loop run` accepts.""" 

120 return str(path.relative_to(base).with_suffix("")) 

121 

122 # Collect project loops (skipped when --builtin is set). Recurse into 

123 # subdirectories (e.g. oracles/) and filter to runnable FSM definitions so 

124 # library fragments under loops/lib/ stay hidden. 

125 project_names: set[str] = set() 

126 yaml_files: list[Path] = [] 

127 if not builtin_only and loops_dir.exists(): 

128 yaml_files = sorted(p for p in loops_dir.rglob("*.yaml") if is_runnable_loop(p)) 

129 project_names = {_rel_key(p, loops_dir) for p in yaml_files} 

130 

131 # Collect built-in loops (excluding those overridden by a project loop at 

132 # the same relative path). 

133 builtin_dir = get_builtin_loops_dir() 

134 builtin_files: list[Path] = [] 

135 if builtin_dir.exists(): 

136 builtin_files = [ 

137 f 

138 for f in sorted(builtin_dir.rglob("*.yaml")) 

139 if is_runnable_loop(f) and _rel_key(f, builtin_dir) not in project_names 

140 ] 

141 

142 if not yaml_files and not builtin_files: 

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

144 print_json([]) 

145 return 0 

146 print("No loops available") 

147 return 0 

148 

149 # Build combined metadata list 

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

151 for path in yaml_files: 

152 meta = _load_loop_meta(path) 

153 all_loops.append( 

154 {"name": _rel_key(path, loops_dir), "path": path, "builtin": False, **meta} 

155 ) 

156 for path in builtin_files: 

157 meta = _load_loop_meta(path) 

158 all_loops.append( 

159 {"name": _rel_key(path, builtin_dir), "path": path, "builtin": True, **meta} 

160 ) 

161 

162 # Apply --category filter 

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

164 if category_filter: 

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

166 

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

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

169 if label_filters: 

170 all_loops = [ 

171 lp 

172 for lp in all_loops 

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

174 ] 

175 

176 if not all_loops: 

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

178 print_json([]) 

179 return 0 

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

181 return 0 

182 

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

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

185 for lp in all_loops: 

186 item: dict[str, Any] = { 

187 "name": lp["name"], 

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

189 "category": lp["category"], 

190 "labels": lp["labels"], 

191 } 

192 if lp["builtin"]: 

193 item["built_in"] = True 

194 items.append(item) 

195 print_json(items) 

196 return 0 

197 

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) 

205 

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") 

210 

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() 

215 

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) 

227 

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")) 

235 

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 

242 

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 "" 

249 

250 print(f" {name_str}{desc_str}{suffix_raw}") 

251 return 0 

252 

253 

254_EVENT_TYPE_WIDTH = 16 # width of "handoff_detected" 

255 

256_ANSI_RE = re.compile(r"\033\[[0-9;]*m") 

257 

258 

259def _strip_ansi(text: str) -> str: 

260 return _ANSI_RE.sub("", text) 

261 

262 

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

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

265 if max_len < 1: 

266 return "" 

267 if len(text) <= max_len: 

268 return text 

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

270 

271 

272def _format_history_event( 

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

274) -> str | None: 

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

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

277 try: 

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

279 except (ValueError, TypeError): 

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

281 

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

283 

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

285 return None 

286 

287 ts_str = colorize(ts, "2") 

288 etype_padded = event_type.ljust(_EVENT_TYPE_WIDTH) 

289 etype_color = "0" 

290 detail = "" 

291 extra_lines: list[str] = [] 

292 

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

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

295 

296 if event_type == "loop_start": 

297 etype_color = "1" 

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

299 

300 elif event_type == "loop_complete": 

301 etype_color = "1" 

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

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

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

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

306 

307 elif event_type == "loop_resume": 

308 etype_color = "1" 

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

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

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

312 

313 elif event_type == "state_enter": 

314 etype_color = "34" 

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

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

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

318 

319 elif event_type == "action_start": 

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

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

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

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

324 first_line = ( 

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

326 if is_prompt 

327 else action 

328 ) 

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

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

331 

332 elif event_type == "action_output": 

333 # Only reached in verbose mode 

334 etype_color = "2" 

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

336 

337 elif event_type == "action_complete": 

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

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

340 if exit_code == 0: 

341 etype_color = "2" 

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

343 else: 

344 etype_color = "38;5;208" 

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

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

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

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

349 if session_jsonl: 

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

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

352 if verbose: 

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

354 if output_preview: 

355 avail_w = width - len(_indent) - 2 

356 preview_text = ( 

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

358 ) 

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

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

361 

362 elif event_type == "evaluate": 

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

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

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

366 if verdict == "yes": 

367 etype_color = "32" 

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

369 else: 

370 etype_color = "38;5;208" 

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

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

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

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

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

376 if verbose: 

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

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

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

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

381 if llm_model or llm_prompt: 

382 meta_parts = [] 

383 if llm_model: 

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

385 if llm_latency_ms != "": 

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

387 meta_str = " ".join(meta_parts) 

388 extra_lines.append( 

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

390 ) 

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

392 if llm_prompt: 

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

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

395 if llm_raw_output: 

396 resp_text = ( 

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

398 ) 

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

400 

401 elif event_type == "route": 

402 etype_color = "2" 

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

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

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

406 

407 elif event_type == "handoff_detected": 

408 etype_color = "33" 

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

410 

411 else: 

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

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

414 

415 etype_str = colorize(etype_padded, etype_color) 

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

417 if extra_lines: 

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

419 return main_line 

420 

421 

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

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

424 if ms < 1000: 

425 return f"{ms}ms" 

426 s = ms // 1000 

427 if s < 60: 

428 return f"{s}s" 

429 m, s = divmod(s, 60) 

430 if m < 60: 

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

432 h, m = divmod(m, 60) 

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

434 

435 

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

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

438 import json as _json 

439 

440 from little_loops.fsm.persistence import HISTORY_DIR, LoopState 

441 

442 history_base = loops_dir / HISTORY_DIR 

443 if not history_base.exists(): 

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

445 return 0 

446 

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

448 suffix = f"-{loop_name}" 

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

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

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

452 continue 

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

454 state_file = run_dir / "state.json" 

455 state: LoopState | None = None 

456 if state_file.exists(): 

457 try: 

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

459 state = LoopState.from_dict(data) 

460 except (ValueError, KeyError): 

461 pass 

462 runs.append((run_id, state)) 

463 

464 if not runs: 

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

466 return 0 

467 

468 if as_json: 

469 print( 

470 _json.dumps( 

471 [ 

472 { 

473 "run_id": rid, 

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

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

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

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

478 } 

479 for rid, s in runs 

480 ], 

481 indent=2, 

482 ) 

483 ) 

484 return 0 

485 

486 status_colors = { 

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

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

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

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

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

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

493 } 

494 reset = "\033[0m" 

495 

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

497 print() 

498 

499 for run_id, state in runs: 

500 if state is not None: 

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

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

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

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

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

506 else: 

507 status_str = "unknown" 

508 duration_str = "?" 

509 started = "?" 

510 iters = "?" 

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

512 

513 print() 

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

515 return 0 

516 

517 

518def cmd_history( 

519 loop_name: str, 

520 run_id: str | None, 

521 args: argparse.Namespace, 

522 loops_dir: Path, 

523) -> int: 

524 """Show loop history. 

525 

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

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

528 """ 

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

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

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

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

533 

534 if run_id is None: 

535 return _list_archived_runs(loop_name, loops_dir, as_json) 

536 

537 # Show events for a specific archived run 

538 from little_loops.fsm.persistence import get_archived_events 

539 

540 events = get_archived_events(loop_name, run_id, loops_dir) 

541 

542 if not events: 

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

544 return 1 

545 

546 w = terminal_width() 

547 if not verbose: 

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

549 

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

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

552 if event_filter: 

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

554 

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

556 if state_filter: 

557 events = [ 

558 e 

559 for e in events 

560 if ( 

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

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

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

564 ) 

565 ] 

566 

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

568 if since_str: 

569 from datetime import timedelta 

570 

571 from little_loops.text_utils import parse_duration 

572 

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

574 events = [ 

575 e 

576 for e in events 

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

578 ] 

579 

580 if as_json: 

581 print_json(events[-tail:]) 

582 return 0 

583 for event in events[-tail:]: 

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

585 if line is not None: 

586 print(line) 

587 

588 return 0 

589 

590 

591def cmd_audit_meta(loop_name: str, args: argparse.Namespace, loops_dir: Path) -> int: 

592 """Summarize meta-eval.jsonl agreement stats from all archived runs of a loop. 

593 

594 Reads meta-eval.jsonl from each archived run, computes: 

595 - Total iterations with llm_structured evaluate events 

596 - Agreement rate (agreed / total) 

597 - Mean diff size (files_changed) per verdict 

598 - Divergence flags: 

599 - agreed=false streak >=3 → "LLM optimistic drift detected" 

600 - agreed=true with files_changed==0 streak >=3 → "Trivial agreement detected" 

601 

602 Returns 0 if no flags triggered, 1 if any threshold crossed. 

603 """ 

604 import json as _json 

605 

606 from little_loops.fsm.persistence import HISTORY_DIR 

607 

608 history_root = loops_dir / HISTORY_DIR 

609 if not history_root.exists(): 

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

611 return 0 

612 

613 suffix = f"-{loop_name}" 

614 all_entries: list[dict[str, Any]] = [] 

615 

616 for run_dir in sorted(history_root.iterdir(), key=lambda d: d.name): 

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

618 continue 

619 meta_eval_file = run_dir / "meta-eval.jsonl" 

620 if not meta_eval_file.exists(): 

621 continue 

622 for line in meta_eval_file.read_text(encoding="utf-8").splitlines(): 

623 line = line.strip() 

624 if line: 

625 try: 

626 all_entries.append(_json.loads(line)) 

627 except _json.JSONDecodeError: 

628 pass 

629 

630 if not all_entries: 

631 print(f"No meta-eval data for: {loop_name}") 

632 return 0 

633 

634 total = len(all_entries) 

635 agreed_count = sum(1 for e in all_entries if e.get("agreed") is True) 

636 agreement_rate = agreed_count / total if total else 0.0 

637 

638 # Mean diff size per verdict 

639 agreed_sizes = [ 

640 e.get("diff_stats", {}).get("files_changed", 0) or 0 

641 for e in all_entries 

642 if e.get("agreed") is True 

643 ] 

644 disagreed_sizes = [ 

645 e.get("diff_stats", {}).get("files_changed", 0) or 0 

646 for e in all_entries 

647 if e.get("agreed") is False 

648 ] 

649 mean_agreed = sum(agreed_sizes) / len(agreed_sizes) if agreed_sizes else 0.0 

650 mean_disagreed = sum(disagreed_sizes) / len(disagreed_sizes) if disagreed_sizes else 0.0 

651 

652 # Detect streaks 

653 flags: list[str] = [] 

654 optimistic_streak = 0 

655 trivial_streak = 0 

656 max_optimistic = 0 

657 max_trivial = 0 

658 

659 for entry in all_entries: 

660 agreed = entry.get("agreed") 

661 files_changed = (entry.get("diff_stats") or {}).get("files_changed", 0) or 0 

662 

663 if agreed is False: 

664 optimistic_streak += 1 

665 trivial_streak = 0 

666 elif agreed is True and files_changed == 0: 

667 trivial_streak += 1 

668 optimistic_streak = 0 

669 else: 

670 optimistic_streak = 0 

671 trivial_streak = 0 

672 

673 max_optimistic = max(max_optimistic, optimistic_streak) 

674 max_trivial = max(max_trivial, trivial_streak) 

675 

676 if max_optimistic >= 3: 

677 flags.append(f"LLM optimistic drift detected (streak={max_optimistic})") 

678 if max_trivial >= 3: 

679 flags.append(f"Trivial agreement detected (streak={max_trivial})") 

680 

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

682 if as_json: 

683 result = { 

684 "loop": loop_name, 

685 "total_entries": total, 

686 "agreed_count": agreed_count, 

687 "agreement_rate": round(agreement_rate, 4), 

688 "mean_files_changed_when_agreed": round(mean_agreed, 2), 

689 "mean_files_changed_when_disagreed": round(mean_disagreed, 2), 

690 "max_optimistic_streak": max_optimistic, 

691 "max_trivial_streak": max_trivial, 

692 "flags": flags, 

693 } 

694 print_json(result) 

695 else: 

696 print(f"Meta-eval audit: {loop_name}") 

697 print(f" Total entries: {total}") 

698 print(f" Agreed: {agreed_count}/{total} ({agreement_rate:.0%})") 

699 print(f" Mean Δfiles agreed: {mean_agreed:.1f}") 

700 print(f" Mean Δfiles disagreed: {mean_disagreed:.1f}") 

701 if flags: 

702 print() 

703 for flag in flags: 

704 print(f" ⚠ {flag}") 

705 else: 

706 print(" No divergence flags.") 

707 

708 if as_json: 

709 return 0 # let caller inspect JSON for flags; non-zero kills provider.runCli 

710 return 1 if flags else 0 

711 

712 

713# --------------------------------------------------------------------------- 

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

715# --------------------------------------------------------------------------- 

716 

717 

718# --------------------------------------------------------------------------- 

719# State overview table 

720# --------------------------------------------------------------------------- 

721 

722 

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

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