Coverage for little_loops / fsm / persistence.py: 19%

1"""State persistence and event streaming for FSM loops. 

2 

3This module provides persistence capabilities for FSM loop execution: 

4- LoopState: Dataclass representing loop execution state 

5- StatePersistence: File I/O for state and events 

6- PersistentExecutor: Wrapper that persists state during execution 

7- Utility functions for listing running loops and reading history 

8 

9File structure: 

10 .loops/ 

11 ├── fix-types.yaml # Loop definition 

12 ├── .running/ # Runtime state (auto-managed) 

13 │ ├── fix-types-20260503T122306.state.json 

14 │ └── fix-types-20260503T122306.events.jsonl 

15 └── .history/ # Archived run logs (auto-populated) 

16 └── 2024-01-15T103000-fix-types/ 

17 ├── state.json 

18 ├── events.jsonl 

19 └── summary.json # present when loop wrote one to run_dir 

20""" 

21 

22from __future__ import annotations 

23 

24import json 

25import logging 

26import os 

27import re 

28import shutil 

29import subprocess 

30import tempfile 

31import time 

32from dataclasses import dataclass, field 

33from datetime import UTC, datetime 

34from pathlib import Path 

35from typing import Any 

36 

37from little_loops.events import EventBus 

38from little_loops.fsm.concurrency import _process_alive 

39from little_loops.fsm.executor import EventCallback, ExecutionResult, FSMExecutor 

40from little_loops.fsm.schema import FSMLoop 

41from little_loops.fsm.validation import _is_meta_loop 

42 

43RUNNING_DIR = ".running" 

44HISTORY_DIR = ".history" 

45 

46RESUMABLE_STATUSES: frozenset[str] = frozenset({"running", "awaiting_continuation", "interrupted"}) 

47 

48logger = logging.getLogger(__name__) 

49 

50_RUN_FOLDER = re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{6})-(.+)$") 

51_INSTANCE_SUFFIX = re.compile(r"-\d{8}T\d{6}$") 

52 

53 

54def _parse_run_folder(name: str) -> tuple[str, str] | None: 

55 """Return (run_id, loop_name) from a flat history folder name, or None.""" 

56 m = _RUN_FOLDER.match(name) 

57 return (m.group(1), m.group(2)) if m else None 

58 

59 

60def _iso_now() -> str: 

61 """Return current time as ISO 8601 string.""" 

62 return datetime.now(UTC).isoformat() 

63 

64 

65def _now_ms() -> int: 

66 """Return current time in milliseconds.""" 

67 return int(time.time() * 1000) 

68 

69 

70def _verdict_is_yes(verdict: str) -> bool: 

71 """Return True if verdict maps to a positive (yes) outcome.""" 

72 return verdict.startswith("yes") or verdict in ("progress", "success") 

73 

74 

75def _parse_diff_stat(text: str) -> dict[str, int] | None: 

76 """Parse 'git diff --stat' summary line into structured dict.""" 

77 last_line = text.strip().rsplit("\n", 1)[-1] if text.strip() else "" 

78 m = re.search( 

79 r"(\d+) files? changed(?:, (\d+) insertions?\(\+\))?(?:, (\d+) deletions?\(-\))?", 

80 last_line, 

81 ) 

82 if not m: 

83 return None 

84 return { 

85 "files_changed": int(m.group(1)), 

86 "insertions": int(m.group(2) or 0), 

87 "deletions": int(m.group(3) or 0), 

88 } 

89 

90 

91def _get_diff_stats() -> dict[str, int] | None: 

92 """Run 'git diff --stat HEAD' and return structured stats, or None on failure.""" 

93 try: 

94 proc = subprocess.run( 

95 ["git", "diff", "--stat", "HEAD"], 

96 capture_output=True, 

97 text=True, 

98 timeout=10, 

99 ) 

100 if proc.returncode == 0: 

101 return _parse_diff_stat(proc.stdout) 

102 except Exception: 

103 pass 

104 return None 

105 

106 

107def _read_pid_file(pid_file: Path) -> int | None: 

108 """Read and validate a PID file, returning the PID or None.""" 

109 if not pid_file.exists(): 

110 return None 

111 try: 

112 return int(pid_file.read_text().strip()) 

113 except (ValueError, OSError): 

114 return None 

115 

116 

117def _resolve_live_pid(running_dir: Path, stem: str, state: LoopState) -> int | None: 

118 """Return the canonical PID for an instance via .pid → .lock → state.pid chain. 

119 

120 Returns None when no PID can be resolved from any source. 

121 """ 

122 pid = _read_pid_file(running_dir / f"{stem}.pid") 

123 if pid is not None: 

124 return pid 

125 lock_file = running_dir / f"{stem}.lock" 

126 if lock_file.exists(): 

127 try: 

128 with open(lock_file) as _lf: 

129 lock_data = json.load(_lf) 

130 pid = lock_data.get("pid") 

131 if pid is not None: 

132 return pid 

133 except (json.JSONDecodeError, KeyError, OSError): 

134 pass 

135 return state.pid 

136 

137 

138def _reconcile_stale_running( 

139 state: LoopState, 

140 persistence: StatePersistence, 

141 running_dir: Path, 

142 stem: str, 

143) -> LoopState: 

144 """Flip a running-state entry to interrupted when its PID is provably dead. 

145 

146 Called on the read path in cmd_status and list_running_loops so orphaned 

147 foreground-crash entries self-heal without requiring manual cleanup-loops 

148 intervention. 

149 """ 

150 if state.status != "running": 

151 return state 

152 pid = _resolve_live_pid(running_dir, stem, state) 

153 if pid is None: 

154 return state # no PID resolvable — cannot determine liveness, leave alone 

155 if _process_alive(pid): 

156 return state 

157 state.status = "interrupted" 

158 state.reconciled_at = datetime.now(UTC).isoformat() 

159 persistence.save_state(state) 

160 return state 

161 

162 

163@dataclass 

164class LoopState: 

165 """Persistent state for an FSM loop execution. 

166 

167 This captures all runtime state needed to resume a loop: 

168 - Current state and iteration 

169 - Captured variables and previous result 

170 - Last evaluation result 

171 - Timestamps and status 

172 

173 Attributes: 

174 loop_name: Name of the loop 

175 current_state: Current FSM state name 

176 iteration: Current iteration count (1-based) 

177 captured: Captured action outputs by variable name 

178 prev_result: Previous state's result (output, exit_code, state) 

179 last_result: Last evaluation result (verdict, details) 

180 started_at: ISO timestamp when loop started 

181 updated_at: ISO timestamp when state was last saved 

182 status: Execution status (running, completed, failed, interrupted, awaiting_continuation, timed_out) 

183 continuation_prompt: Continuation context from handoff signal (if status is awaiting_continuation) 

184 accumulated_ms: Total milliseconds elapsed across all segments up to this save (used to restore 

185 elapsed time correctly after resume, so duration_ms and ${loop.elapsed_ms} reflect the 

186 full loop lifetime rather than only the most recent segment) 

187 """ 

188 

189 loop_name: str 

190 current_state: str 

191 iteration: int 

192 captured: dict[str, dict[str, Any]] 

193 prev_result: dict[str, Any] | None 

194 last_result: dict[str, Any] | None 

195 started_at: str 

196 updated_at: str 

197 status: ( 

198 str # "running", "completed", "failed", "interrupted", "awaiting_continuation", "timed_out" 

199 ) 

200 continuation_prompt: str | None = None 

201 accumulated_ms: int = 0 # total elapsed ms across all segments (for resume offset) 

202 retry_counts: dict[str, int] = field(default_factory=dict) # per-state retry tracking 

203 # Per-state rate-limit retry tracking (ENH-1133: dict-of-record). 

204 # Each record: {"short_retries": int, "long_retries": int, 

205 # "total_wait_seconds": float, "first_seen_at": float | None}. 

206 # Legacy int values (dict[str, int]) are coerced in from_dict. 

207 rate_limit_retries: dict[str, dict[str, Any]] = field(default_factory=dict) 

208 # Count of consecutive rate_limit_exhausted emissions across states. Reset 

209 # on any non-rate-limited state outcome. Persisted for resume durability. 

210 consecutive_rate_limit_exhaustions: int = 0 

211 # Per-edge revisit tracking for cycle detection. 

212 edge_revisit_counts: dict[str, int] = field(default_factory=dict) 

213 # BUG-2204: full-pass counter (maintain-mode restarts); 0 for loops without maintain. 

214 iteration_count: int = 0 

215 active_sub_loop: str | None = None # name of currently executing sub-loop (observability) 

216 pid: int | None = None # OS PID of the process that started this run (for reconciliation sweep) 

217 reconciled_at: str | None = None # ISO timestamp when orphaned-running state was auto-flipped 

218 messages: list[str] = field(default_factory=list) 

219 

220 def to_dict(self) -> dict[str, Any]: 

221 """Convert to dictionary for JSON serialization.""" 

222 result = { 

223 "loop_name": self.loop_name, 

224 "current_state": self.current_state, 

225 "iteration": self.iteration, 

226 "captured": self.captured, 

227 "prev_result": self.prev_result, 

228 "last_result": self.last_result, 

229 "started_at": self.started_at, 

230 "updated_at": self.updated_at, 

231 "status": self.status, 

232 "accumulated_ms": self.accumulated_ms, 

233 } 

234 if self.continuation_prompt is not None: 

235 result["continuation_prompt"] = self.continuation_prompt 

236 if self.retry_counts: 

237 result["retry_counts"] = self.retry_counts 

238 if self.rate_limit_retries: 

239 result["rate_limit_retries"] = self.rate_limit_retries 

240 if self.consecutive_rate_limit_exhaustions: 

241 result["consecutive_rate_limit_exhaustions"] = self.consecutive_rate_limit_exhaustions 

242 if self.edge_revisit_counts: 

243 result["edge_revisit_counts"] = self.edge_revisit_counts 

244 if self.iteration_count: 

245 result["iteration_count"] = self.iteration_count 

246 if self.active_sub_loop is not None: 

247 result["active_sub_loop"] = self.active_sub_loop 

248 if self.pid is not None: 

249 result["pid"] = self.pid 

250 if self.reconciled_at is not None: 

251 result["reconciled_at"] = self.reconciled_at 

252 if self.messages: 

253 result["messages"] = self.messages 

254 return result 

255 

256 @classmethod 

257 def from_dict(cls, data: dict[str, Any]) -> LoopState: 

258 """Create LoopState from dictionary. 

259 

260 Migrates legacy ``rate_limit_retries`` values from ``dict[str, int]`` 

261 (BUG-1107 pre-ENH-1133 shape) to the dict-of-record shape. Integer 

262 values are coerced to ``{"short_retries": <int>, "long_retries": 0, 

263 "total_wait_seconds": 0.0, "first_seen_at": None}``. 

264 

265 Args: 

266 data: Dictionary with loop state fields 

267 

268 Returns: 

269 LoopState instance 

270 """ 

271 raw_rl = data.get("rate_limit_retries", {}) or {} 

272 migrated_rl: dict[str, dict[str, Any]] = {} 

273 for state_name, value in raw_rl.items(): 

274 if isinstance(value, int): 

275 migrated_rl[state_name] = { 

276 "short_retries": value, 

277 "long_retries": 0, 

278 "total_wait_seconds": 0.0, 

279 "first_seen_at": None, 

280 } 

281 elif isinstance(value, dict): 

282 migrated_rl[state_name] = value 

283 return cls( 

284 loop_name=data["loop_name"], 

285 current_state=data["current_state"], 

286 iteration=data["iteration"], 

287 captured=data.get("captured", {}), 

288 prev_result=data.get("prev_result"), 

289 last_result=data.get("last_result"), 

290 started_at=data["started_at"], 

291 updated_at=data.get("updated_at", ""), 

292 status=data["status"], 

293 continuation_prompt=data.get("continuation_prompt"), 

294 accumulated_ms=data.get("accumulated_ms", 0), 

295 retry_counts=data.get("retry_counts", {}), 

296 rate_limit_retries=migrated_rl, 

297 consecutive_rate_limit_exhaustions=data.get("consecutive_rate_limit_exhaustions", 0), 

298 edge_revisit_counts=data.get("edge_revisit_counts", {}), 

299 iteration_count=data.get("iteration_count", 0), 

300 active_sub_loop=data.get("active_sub_loop"), 

301 pid=data.get("pid"), 

302 reconciled_at=data.get("reconciled_at"), 

303 messages=data.get("messages", []), 

304 ) 

305 

306 

307class StatePersistence: 

308 """Manage loop state persistence and event streaming. 

309 

310 Handles file I/O for: 

311 - State file: JSON file with current execution state 

312 - Events file: JSONL file with execution events (append-only) 

313 

314 Files are stored in .loops/.running/<instance_id>.* 

315 """ 

316 

317 def __init__( 

318 self, loop_name: str, loops_dir: Path | None = None, instance_id: str | None = None 

319 ) -> None: 

320 """Initialize persistence for a loop. 

321 

322 Args: 

323 loop_name: Name of the loop 

324 loops_dir: Base directory for loops (default: .loops) 

325 instance_id: Optional unique instance identifier; falls back to loop_name when None 

326 """ 

327 self.loop_name = loop_name 

328 self.loops_dir = loops_dir or Path(".loops") 

329 self.running_dir = self.loops_dir / RUNNING_DIR 

330 stem = instance_id or loop_name 

331 self.state_file = self.running_dir / f"{stem}.state.json" 

332 self.events_file = self.running_dir / f"{stem}.events.jsonl" 

333 self.meta_eval_file = self.running_dir / f"{stem}.meta-eval.jsonl" 

334 

335 def initialize(self) -> None: 

336 """Create running directory if needed.""" 

337 self.running_dir.mkdir(parents=True, exist_ok=True) 

338 

339 def save_state(self, state: LoopState) -> None: 

340 """Save current state to file using an atomic write. 

341 

342 Updates the updated_at timestamp before saving. Writes to a temporary 

343 file first, then renames it over the target to avoid leaving a corrupt 

344 or empty state file if the process is killed mid-write. 

345 

346 Args: 

347 state: LoopState to save 

348 """ 

349 state.updated_at = _iso_now() 

350 data = json.dumps(state.to_dict(), indent=2) 

351 tmp_fd, tmp_path = tempfile.mkstemp(dir=self.state_file.parent, suffix=".tmp") 

352 try: 

353 with os.fdopen(tmp_fd, "w") as f: 

354 f.write(data) 

355 os.replace(tmp_path, self.state_file) 

356 except Exception: 

357 os.unlink(tmp_path) 

358 raise 

359 

360 def load_state(self) -> LoopState | None: 

361 """Load state from file, or None if not exists. 

362 

363 Returns: 

364 LoopState if file exists and is valid, None otherwise 

365 """ 

366 if not self.state_file.exists(): 

367 return None 

368 try: 

369 data = json.loads(self.state_file.read_text()) 

370 except json.JSONDecodeError: 

371 return None 

372 try: 

373 return LoopState.from_dict(data) 

374 except KeyError as e: 

375 logger.warning("Corrupted state file %s: missing key %s", self.state_file, e) 

376 return None 

377 

378 def clear_state(self) -> None: 

379 """Remove state file.""" 

380 if self.state_file.exists(): 

381 self.state_file.unlink() 

382 

383 def append_event(self, event: dict[str, Any]) -> None: 

384 """Append event to JSONL file. 

385 

386 Args: 

387 event: Event dictionary to append 

388 """ 

389 with open(self.events_file, "a", encoding="utf-8") as f: 

390 f.write(json.dumps(event) + "\n") 

391 

392 def read_events(self) -> list[dict[str, Any]]: 

393 """Read all events from file. 

394 

395 Returns: 

396 List of event dictionaries, empty if file doesn't exist 

397 """ 

398 if not self.events_file.exists(): 

399 return [] 

400 events: list[dict[str, Any]] = [] 

401 with open(self.events_file, encoding="utf-8") as f: 

402 for line in f: 

403 line = line.strip() 

404 if line: 

405 try: 

406 events.append(json.loads(line)) 

407 except json.JSONDecodeError: 

408 continue # Skip malformed lines 

409 return events 

410 

411 def clear_events(self) -> None: 

412 """Remove events file.""" 

413 if self.events_file.exists(): 

414 self.events_file.unlink() 

415 

416 def clear_meta_eval(self) -> None: 

417 """Remove meta-eval file.""" 

418 if self.meta_eval_file.exists(): 

419 self.meta_eval_file.unlink() 

420 

421 def archive_run(self, run_dir: Path | None = None) -> Path | None: 

422 """Archive current run files to .history/ before clearing. 

423 

424 Reads the current state to derive the run timestamp, then copies 

425 state.json, events.jsonl, and (when present) meta-eval.jsonl and 

426 summary.json into: 

427 <loops_dir>/.history/<run_id>-<loop_name>/ 

428 

429 where run_id is a compact ISO timestamp derived from started_at 

430 (e.g. "2024-01-15T103000" from "2024-01-15T10:30:00.123456+00:00"). 

431 

432 Args: 

433 run_dir: Optional path to the loop's run directory. When provided, 

434 summary.json is copied from run_dir to the archive directory if 

435 it exists. Pass None (default) when the run directory is not 

436 available (e.g. stale-run cleanup paths). 

437 

438 Returns: 

439 Path to the archive directory if files were archived, None if 

440 there were no files to archive (fresh run). 

441 """ 

442 has_state = self.state_file.exists() 

443 has_events = self.events_file.exists() 

444 if not has_state and not has_events: 

445 return None 

446 

447 # Derive run ID from started_at in state file, or fall back to now 

448 state = self.load_state() 

449 if state is not None and state.started_at: 

450 # Compact ISO: strip colons, dots, plus signs; take first 19 chars 

451 # e.g. "2024-01-15T10:30:00.123+00:00" → "2024-01-15T103000" 

452 run_id = state.started_at.replace(":", "").replace(".", "").replace("+", "")[:17] 

453 else: 

454 run_id = datetime.now(UTC).strftime("%Y-%m-%dT%H%M%S") 

455 

456 run_folder = f"{run_id}-{self.loop_name}" 

457 archive_dir = self.loops_dir / HISTORY_DIR / run_folder 

458 archive_dir.mkdir(parents=True, exist_ok=True) 

459 

460 if has_state: 

461 shutil.copy2(self.state_file, archive_dir / "state.json") 

462 if has_events: 

463 shutil.copy2(self.events_file, archive_dir / "events.jsonl") 

464 if self.meta_eval_file.exists(): 

465 shutil.copy2(self.meta_eval_file, archive_dir / "meta-eval.jsonl") 

466 if run_dir is not None: 

467 summary_src = run_dir / "summary.json" 

468 if summary_src.exists(): 

469 shutil.copy2(summary_src, archive_dir / "summary.json") 

470 

471 return archive_dir 

472 

473 def clear_all(self) -> None: 

474 """Archive current run files then clear state and events (for new run).""" 

475 self.archive_run() 

476 self.clear_state() 

477 self.clear_events() 

478 self.clear_meta_eval() 

479 

480 

481def _reconcile_stale_runs(loops_dir: Path) -> int: 

482 """Archive state files in .running/ that belong to dead or terminal processes. 

483 

484 Called at loop startup to clean up files left by crashed or interrupted runs. 

485 Returns the count of archived files. 

486 

487 Strategy (mirrors LockManager.find_conflict() stale-lock cleanup): 

488 - Terminal-status files (completed/failed/timed_out) are archived 

489 unconditionally — they are definitionally stale by invariant. 

490 - status="interrupted" files are left alone so the user can resume them. 

491 - status="running" files are checked via their sibling .pid file; archived 

492 only if the PID is confirmed dead. No .pid file → leave alone (can't confirm). 

493 """ 

494 running_dir = loops_dir / RUNNING_DIR 

495 if not running_dir.exists(): 

496 return 0 

497 

498 terminal_statuses = {"completed", "failed", "timed_out"} 

499 archived = 0 

500 

501 for state_file in running_dir.glob("*.state.json"): 

502 try: 

503 data = json.loads(state_file.read_text()) 

504 state = LoopState.from_dict(data) 

505 except (json.JSONDecodeError, KeyError, OSError): 

506 continue 

507 

508 is_stale = state.status in terminal_statuses 

509 

510 if not is_stale and state.status == "running": 

511 stem = state_file.name.removesuffix(".state.json") 

512 pid_file = running_dir / f"{stem}.pid" 

513 if pid_file.exists(): 

514 try: 

515 pid = int(pid_file.read_text().strip()) 

516 is_stale = not _process_alive(pid) 

517 except (OSError, ValueError): 

518 pass 

519 

520 if not is_stale: 

521 continue 

522 

523 stem = state_file.name.removesuffix(".state.json") 

524 instance_id = stem if stem != state.loop_name else None 

525 persistence = StatePersistence( 

526 loop_name=state.loop_name, 

527 loops_dir=loops_dir, 

528 instance_id=instance_id, 

529 ) 

530 try: 

531 persistence.clear_all() 

532 (running_dir / f"{stem}.pid").unlink(missing_ok=True) 

533 archived += 1 

534 logger.debug("Archived stale run: %s (status=%s)", stem, state.status) 

535 except OSError as e: 

536 logger.warning("Failed to archive stale run %s: %s", stem, e) 

537 

538 if archived: 

539 logger.info("Reconciliation sweep archived %d stale run(s) from .running/", archived) 

540 

541 return archived 

542 

543 

544class PersistentExecutor: 

545 """FSM Executor with state persistence and event streaming. 

546 

547 Wraps FSMExecutor to: 

548 - Save state after each state transition 

549 - Append events to JSONL file as they occur 

550 - Support resuming from saved state 

551 - Support graceful shutdown via signal handling 

552 """ 

553 

554 def __init__( 

555 self, 

556 fsm: FSMLoop, 

557 persistence: StatePersistence | None = None, 

558 loops_dir: Path | None = None, 

559 instance_id: str | None = None, 

560 pid: int | None = None, 

561 **executor_kwargs: Any, 

562 ) -> None: 

563 """Initialize persistent executor. 

564 

565 Args: 

566 fsm: FSM loop definition 

567 persistence: Optional pre-configured persistence (for testing) 

568 loops_dir: Base directory for loops (default: .loops) 

569 instance_id: Optional unique instance identifier for file path scoping 

570 pid: OS PID of the running process; stored in saved state for reconciliation 

571 **executor_kwargs: Additional kwargs for FSMExecutor 

572 """ 

573 from little_loops.fsm.handoff_handler import HandoffBehavior, HandoffHandler 

574 from little_loops.fsm.signal_detector import SignalDetector 

575 

576 self.fsm = fsm 

577 self.loops_dir = loops_dir 

578 self._run_pid = pid 

579 self.persistence = persistence or StatePersistence( 

580 fsm.name, loops_dir or Path(".loops"), instance_id=instance_id 

581 ) 

582 self.persistence.initialize() 

583 

584 # Create signal detector and handler based on FSM config 

585 signal_detector = SignalDetector() 

586 handoff_handler = HandoffHandler(HandoffBehavior(fsm.on_handoff)) 

587 

588 # Create base executor with event callback that persists 

589 self._executor = FSMExecutor( 

590 fsm, 

591 event_callback=self._handle_event, 

592 signal_detector=signal_detector, 

593 handoff_handler=handoff_handler, 

594 loops_dir=self.loops_dir, 

595 **executor_kwargs, 

596 ) 

597 self._last_result: dict[str, Any] | None = None 

598 self._last_non_llm_result: dict[str, Any] | None = None 

599 self._continuation_prompt: str | None = None 

600 self.event_bus = EventBus() 

601 

602 @property 

603 def _on_event(self) -> EventCallback | None: 

604 """Backward-compatible access to the first observer on the event bus.""" 

605 return self.event_bus._observers[0][0] if self.event_bus._observers else None 

606 

607 @_on_event.setter 

608 def _on_event(self, callback: EventCallback | None) -> None: 

609 """Backward-compatible setter: replaces all observers with this one.""" 

610 self.event_bus._observers.clear() 

611 if callback is not None: 

612 self.event_bus.register(callback) 

613 

614 def close_transports(self) -> None: 

615 """Close all transports registered on the underlying EventBus.""" 

616 self.event_bus.close_transports() 

617 

618 def request_shutdown(self) -> None: 

619 """Request graceful shutdown of the executor. 

620 

621 Delegates to the underlying FSMExecutor's request_shutdown method. 

622 The loop will exit cleanly after the current state completes, 

623 saving state as "interrupted" so it can be resumed later. 

624 """ 

625 self._executor.request_shutdown() 

626 

627 def _handle_event(self, event: dict[str, Any]) -> None: 

628 """Handle event: persist to file and save state. 

629 

630 Args: 

631 event: Event dictionary from executor 

632 """ 

633 self.persistence.append_event(event) 

634 

635 event_type = event.get("event") 

636 

637 # Write per-state token usage to usage.jsonl when an LLM action completes. 

638 # Shell and mcp_tool invocations produce no token data and are skipped. 

639 if event_type == "action_complete" and "input_tokens" in event: 

640 run_dir = self.fsm.context.get("run_dir", "") 

641 if run_dir: 

642 usage_path = Path(run_dir) / "usage.jsonl" 

643 entry = { 

644 "iteration": self._executor.iteration, 

645 "state": self._executor.current_state, 

646 "action_type": "prompt" if event.get("is_prompt") else "shell_or_mcp", 

647 "input_tokens": event["input_tokens"], 

648 "output_tokens": event["output_tokens"], 

649 "cache_read_tokens": event.get("cache_read_tokens", 0), 

650 "cache_creation_tokens": event.get("cache_creation_tokens", 0), 

651 "model": event.get("model", "unknown"), 

652 "timestamp": event.get("ts", ""), 

653 } 

654 with open(usage_path, "a", encoding="utf-8") as f: 

655 f.write(json.dumps(entry) + "\n") 

656 

657 # Append shared message to messages.jsonl when a state appends to the log. 

658 if event_type == "messages_append": 

659 run_dir = self.fsm.context.get("run_dir", "") 

660 if run_dir: 

661 messages_path = Path(run_dir) / "messages.jsonl" 

662 entry = { 

663 "iteration": self._executor.iteration, 

664 "state": event.get("state", ""), 

665 "message": event.get("message", ""), 

666 "timestamp": event.get("ts", ""), 

667 } 

668 with open(messages_path, "a", encoding="utf-8") as f: 

669 f.write(json.dumps(entry) + "\n") 

670 

671 # Save state after state transitions 

672 if event_type in ("state_enter", "loop_complete", "baseline_complete"): 

673 self._save_state() 

674 

675 # Track evaluation results for state persistence 

676 if event_type == "evaluate": 

677 self._last_result = { 

678 "verdict": event.get("verdict"), 

679 "details": { 

680 k: v for k, v in event.items() if k not in ("event", "ts", "type", "verdict") 

681 }, 

682 } 

683 eval_type = event.get("type", "") 

684 if eval_type != "llm_structured": 

685 self._last_non_llm_result = { 

686 "state": self._executor.current_state, 

687 "evaluator": eval_type, 

688 "verdict": event.get("verdict", ""), 

689 "details": { 

690 k: v 

691 for k, v in event.items() 

692 if k not in ("event", "ts", "type", "verdict") 

693 }, 

694 } 

695 elif _is_meta_loop(self.fsm): 

696 self._write_meta_eval_entry(event) 

697 

698 # Track handoff events for continuation prompt 

699 if event_type == "handoff_detected": 

700 self._continuation_prompt = event.get("continuation") 

701 

702 # Delegate to registered observers (e.g. progress display, extensions) 

703 self.event_bus.emit(event) 

704 

705 def _write_meta_eval_entry(self, event: dict[str, Any]) -> None: 

706 """Append one JSONL entry to meta-eval.jsonl for an llm_structured evaluate in a meta-loop.""" 

707 non_llm = self._last_non_llm_result or {} 

708 non_llm_details = non_llm.get("details", {}) 

709 

710 llm_verdict = event.get("verdict", "") 

711 ext_verdict = non_llm.get("verdict", "") 

712 agreed: bool | None = ( 

713 _verdict_is_yes(llm_verdict) == _verdict_is_yes(ext_verdict) if ext_verdict else None 

714 ) 

715 

716 ext_value = non_llm_details.get("current", non_llm_details.get("value")) 

717 ext_target = non_llm_details.get("target") 

718 

719 entry: dict[str, Any] = { 

720 "iteration": self._executor.iteration, 

721 "ts": _iso_now(), 

722 "loop": self.fsm.name, 

723 "state": self._executor.current_state, 

724 "llm_verdict": llm_verdict, 

725 "llm_rationale": (event.get("reason") or "")[:200], 

726 "external_verdict": ext_verdict or None, 

727 "external_state": non_llm.get("state"), 

728 "external_evaluator": non_llm.get("evaluator") or None, 

729 "external_value": str(ext_value) if ext_value is not None else None, 

730 "external_target": str(ext_target) if ext_target is not None else None, 

731 "diff_stats": _get_diff_stats(), 

732 "agreed": agreed, 

733 } 

734 with open(self.persistence.meta_eval_file, "a", encoding="utf-8") as f: 

735 f.write(json.dumps(entry) + "\n") 

736 

737 def _save_state(self) -> None: 

738 """Save current executor state to file.""" 

739 status = "running" 

740 if self._executor.current_state: 

741 state_config = self.fsm.states.get(self._executor.current_state) 

742 if state_config and state_config.terminal: 

743 status = "completed" 

744 

745 state = LoopState( 

746 loop_name=self.fsm.name, 

747 current_state=self._executor.current_state, 

748 iteration=self._executor.iteration, 

749 captured=self._executor.captured, 

750 prev_result=self._executor.prev_result, 

751 last_result=self._last_result, 

752 started_at=self._executor.started_at, 

753 updated_at="", # Will be set by save_state 

754 status=status, 

755 accumulated_ms=_now_ms() 

756 - self._executor.start_time_ms 

757 + self._executor.elapsed_offset_ms, 

758 retry_counts=dict(self._executor._retry_counts), 

759 rate_limit_retries={k: dict(v) for k, v in self._executor._rate_limit_retries.items()}, 

760 consecutive_rate_limit_exhaustions=(self._executor._consecutive_rate_limit_exhaustions), 

761 edge_revisit_counts=dict(self._executor._edge_revisit_counts), 

762 iteration_count=self._executor._iteration_count, 

763 pid=self._run_pid, 

764 messages=list(self._executor.messages), 

765 ) 

766 self.persistence.save_state(state) 

767 

768 def run(self, clear_previous: bool = True) -> ExecutionResult: 

769 """Run the FSM with persistence. 

770 

771 Args: 

772 clear_previous: If True, clear previous state/events before running 

773 

774 Returns: 

775 ExecutionResult from the execution 

776 """ 

777 if clear_previous: 

778 self.persistence.clear_all() 

779 

780 result = self._executor.run() 

781 

782 # Update final state 

783 final_status = "completed" if result.terminated_by == "terminal" else "failed" 

784 if result.terminated_by in ("max_steps", "max_iterations_reached", "signal"): 

785 final_status = "interrupted" 

786 if result.terminated_by == "handoff": 

787 final_status = "awaiting_continuation" 

788 if result.terminated_by == "timeout": 

789 final_status = "timed_out" 

790 if result.terminated_by == "cycle_detected": 

791 final_status = "failed" 

792 

793 final_state = LoopState( 

794 loop_name=self.fsm.name, 

795 current_state=result.final_state, 

796 iteration=result.iterations, 

797 captured=result.captured, 

798 prev_result=self._executor.prev_result, 

799 last_result=self._last_result, 

800 started_at=self._executor.started_at, 

801 updated_at="", 

802 status=final_status, 

803 continuation_prompt=self._continuation_prompt, 

804 accumulated_ms=result.duration_ms, 

805 ) 

806 self.persistence.save_state(final_state) 

807 run_dir_str = self.fsm.context.get("run_dir", "")