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

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

20 

21from __future__ import annotations 

22 

23import json 

24import logging 

25import os 

26import re 

27import shutil 

28import subprocess 

29import tempfile 

30import time 

31from dataclasses import dataclass, field 

32from datetime import UTC, datetime 

33from pathlib import Path 

34from typing import Any 

35 

36from little_loops.events import EventBus 

37from little_loops.fsm.concurrency import _process_alive 

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

39from little_loops.fsm.schema import FSMLoop 

40from little_loops.fsm.validation import _is_meta_loop 

41 

42RUNNING_DIR = ".running" 

43HISTORY_DIR = ".history" 

44 

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

46 

47logger = logging.getLogger(__name__) 

48 

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

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

51 

52 

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

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

55 m = _RUN_FOLDER.match(name) 

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

57 

58 

59def _iso_now() -> str: 

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

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

62 

63 

64def _now_ms() -> int: 

65 """Return current time in milliseconds.""" 

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

67 

68 

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

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

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

72 

73 

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

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

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

77 m = re.search( 

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

79 last_line, 

80 ) 

81 if not m: 

82 return None 

83 return { 

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

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

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

87 } 

88 

89 

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

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

92 try: 

93 proc = subprocess.run( 

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

95 capture_output=True, 

96 text=True, 

97 timeout=10, 

98 ) 

99 if proc.returncode == 0: 

100 return _parse_diff_stat(proc.stdout) 

101 except Exception: 

102 pass 

103 return None 

104 

105 

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

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

108 if not pid_file.exists(): 

109 return None 

110 try: 

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

112 except (ValueError, OSError): 

113 return None 

114 

115 

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

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

118 

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

120 """ 

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

122 if pid is not None: 

123 return pid 

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

125 if lock_file.exists(): 

126 try: 

127 with open(lock_file) as _lf: 

128 lock_data = json.load(_lf) 

129 pid = lock_data.get("pid") 

130 if pid is not None: 

131 return pid 

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

133 pass 

134 return state.pid 

135 

136 

137def _reconcile_stale_running( 

138 state: LoopState, 

139 persistence: StatePersistence, 

140 running_dir: Path, 

141 stem: str, 

142) -> LoopState: 

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

144 

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

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

147 intervention. 

148 """ 

149 if state.status != "running": 

150 return state 

151 pid = _resolve_live_pid(running_dir, stem, state) 

152 if pid is None: 

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

154 if _process_alive(pid): 

155 return state 

156 state.status = "interrupted" 

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

158 persistence.save_state(state) 

159 return state 

160 

161 

162@dataclass 

163class LoopState: 

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

165 

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

167 - Current state and iteration 

168 - Captured variables and previous result 

169 - Last evaluation result 

170 - Timestamps and status 

171 

172 Attributes: 

173 loop_name: Name of the loop 

174 current_state: Current FSM state name 

175 iteration: Current iteration count (1-based) 

176 captured: Captured action outputs by variable name 

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

178 last_result: Last evaluation result (verdict, details) 

179 started_at: ISO timestamp when loop started 

180 updated_at: ISO timestamp when state was last saved 

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

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

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

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

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

186 """ 

187 

188 loop_name: str 

189 current_state: str 

190 iteration: int 

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

192 prev_result: dict[str, Any] | None 

193 last_result: dict[str, Any] | None 

194 started_at: str 

195 updated_at: str 

196 status: ( 

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

198 ) 

199 continuation_prompt: str | None = None 

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

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

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

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

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

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

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

207 # Count of consecutive rate_limit_exhausted emissions across states. Reset 

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

209 consecutive_rate_limit_exhaustions: int = 0 

210 # Per-edge revisit tracking for cycle detection. 

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

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

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

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

215 

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

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

218 result = { 

219 "loop_name": self.loop_name, 

220 "current_state": self.current_state, 

221 "iteration": self.iteration, 

222 "captured": self.captured, 

223 "prev_result": self.prev_result, 

224 "last_result": self.last_result, 

225 "started_at": self.started_at, 

226 "updated_at": self.updated_at, 

227 "status": self.status, 

228 "accumulated_ms": self.accumulated_ms, 

229 } 

230 if self.continuation_prompt is not None: 

231 result["continuation_prompt"] = self.continuation_prompt 

232 if self.retry_counts: 

233 result["retry_counts"] = self.retry_counts 

234 if self.rate_limit_retries: 

235 result["rate_limit_retries"] = self.rate_limit_retries 

236 if self.consecutive_rate_limit_exhaustions: 

237 result["consecutive_rate_limit_exhaustions"] = self.consecutive_rate_limit_exhaustions 

238 if self.edge_revisit_counts: 

239 result["edge_revisit_counts"] = self.edge_revisit_counts 

240 if self.active_sub_loop is not None: 

241 result["active_sub_loop"] = self.active_sub_loop 

242 if self.pid is not None: 

243 result["pid"] = self.pid 

244 if self.reconciled_at is not None: 

245 result["reconciled_at"] = self.reconciled_at 

246 return result 

247 

248 @classmethod 

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

250 """Create LoopState from dictionary. 

251 

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

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

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

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

256 

257 Args: 

258 data: Dictionary with loop state fields 

259 

260 Returns: 

261 LoopState instance 

262 """ 

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

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

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

266 if isinstance(value, int): 

267 migrated_rl[state_name] = { 

268 "short_retries": value, 

269 "long_retries": 0, 

270 "total_wait_seconds": 0.0, 

271 "first_seen_at": None, 

272 } 

273 elif isinstance(value, dict): 

274 migrated_rl[state_name] = value 

275 return cls( 

276 loop_name=data["loop_name"], 

277 current_state=data["current_state"], 

278 iteration=data["iteration"], 

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

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

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

282 started_at=data["started_at"], 

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

284 status=data["status"], 

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

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

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

288 rate_limit_retries=migrated_rl, 

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

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

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

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

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

294 ) 

295 

296 

297class StatePersistence: 

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

299 

300 Handles file I/O for: 

301 - State file: JSON file with current execution state 

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

303 

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

305 """ 

306 

307 def __init__( 

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

309 ) -> None: 

310 """Initialize persistence for a loop. 

311 

312 Args: 

313 loop_name: Name of the loop 

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

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

316 """ 

317 self.loop_name = loop_name 

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

319 self.running_dir = self.loops_dir / RUNNING_DIR 

320 stem = instance_id or loop_name 

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

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

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

324 

325 def initialize(self) -> None: 

326 """Create running directory if needed.""" 

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

328 

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

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

331 

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

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

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

335 

336 Args: 

337 state: LoopState to save 

338 """ 

339 state.updated_at = _iso_now() 

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

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

342 try: 

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

344 f.write(data) 

345 os.replace(tmp_path, self.state_file) 

346 except Exception: 

347 os.unlink(tmp_path) 

348 raise 

349 

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

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

352 

353 Returns: 

354 LoopState if file exists and is valid, None otherwise 

355 """ 

356 if not self.state_file.exists(): 

357 return None 

358 try: 

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

360 except json.JSONDecodeError: 

361 return None 

362 try: 

363 return LoopState.from_dict(data) 

364 except KeyError as e: 

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

366 return None 

367 

368 def clear_state(self) -> None: 

369 """Remove state file.""" 

370 if self.state_file.exists(): 

371 self.state_file.unlink() 

372 

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

374 """Append event to JSONL file. 

375 

376 Args: 

377 event: Event dictionary to append 

378 """ 

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

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

381 

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

383 """Read all events from file. 

384 

385 Returns: 

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

387 """ 

388 if not self.events_file.exists(): 

389 return [] 

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

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

392 for line in f: 

393 line = line.strip() 

394 if line: 

395 try: 

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

397 except json.JSONDecodeError: 

398 continue # Skip malformed lines 

399 return events 

400 

401 def clear_events(self) -> None: 

402 """Remove events file.""" 

403 if self.events_file.exists(): 

404 self.events_file.unlink() 

405 

406 def clear_meta_eval(self) -> None: 

407 """Remove meta-eval file.""" 

408 if self.meta_eval_file.exists(): 

409 self.meta_eval_file.unlink() 

410 

411 def archive_run(self) -> Path | None: 

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

413 

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

415 both state.json and events.jsonl into: 

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

417 

418 where run_id is a compact ISO timestamp derived from started_at 

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

420 

421 Returns: 

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

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

424 """ 

425 has_state = self.state_file.exists() 

426 has_events = self.events_file.exists() 

427 if not has_state and not has_events: 

428 return None 

429 

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

431 state = self.load_state() 

432 if state is not None and state.started_at: 

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

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

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

436 else: 

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

438 

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

440 archive_dir = self.loops_dir / HISTORY_DIR / run_folder 

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

442 

443 if has_state: 

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

445 if has_events: 

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

447 if self.meta_eval_file.exists(): 

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

449 

450 return archive_dir 

451 

452 def clear_all(self) -> None: 

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

454 self.archive_run() 

455 self.clear_state() 

456 self.clear_events() 

457 self.clear_meta_eval() 

458 

459 

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

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

462 

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

464 Returns the count of archived files. 

465 

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

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

468 unconditionally — they are definitionally stale by invariant. 

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

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

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

472 """ 

473 running_dir = loops_dir / RUNNING_DIR 

474 if not running_dir.exists(): 

475 return 0 

476 

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

478 archived = 0 

479 

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

481 try: 

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

483 state = LoopState.from_dict(data) 

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

485 continue 

486 

487 is_stale = state.status in terminal_statuses 

488 

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

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

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

492 if pid_file.exists(): 

493 try: 

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

495 is_stale = not _process_alive(pid) 

496 except (OSError, ValueError): 

497 pass 

498 

499 if not is_stale: 

500 continue 

501 

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

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

504 persistence = StatePersistence( 

505 loop_name=state.loop_name, 

506 loops_dir=loops_dir, 

507 instance_id=instance_id, 

508 ) 

509 try: 

510 persistence.clear_all() 

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

512 archived += 1 

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

514 except OSError as e: 

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

516 

517 if archived: 

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

519 

520 return archived 

521 

522 

523class PersistentExecutor: 

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

525 

526 Wraps FSMExecutor to: 

527 - Save state after each state transition 

528 - Append events to JSONL file as they occur 

529 - Support resuming from saved state 

530 - Support graceful shutdown via signal handling 

531 """ 

532 

533 def __init__( 

534 self, 

535 fsm: FSMLoop, 

536 persistence: StatePersistence | None = None, 

537 loops_dir: Path | None = None, 

538 instance_id: str | None = None, 

539 pid: int | None = None, 

540 **executor_kwargs: Any, 

541 ) -> None: 

542 """Initialize persistent executor. 

543 

544 Args: 

545 fsm: FSM loop definition 

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

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

548 instance_id: Optional unique instance identifier for file path scoping 

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

550 **executor_kwargs: Additional kwargs for FSMExecutor 

551 """ 

552 from little_loops.fsm.handoff_handler import HandoffBehavior, HandoffHandler 

553 from little_loops.fsm.signal_detector import SignalDetector 

554 

555 self.fsm = fsm 

556 self.loops_dir = loops_dir 

557 self._run_pid = pid 

558 self.persistence = persistence or StatePersistence( 

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

560 ) 

561 self.persistence.initialize() 

562 

563 # Create signal detector and handler based on FSM config 

564 signal_detector = SignalDetector() 

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

566 

567 # Create base executor with event callback that persists 

568 self._executor = FSMExecutor( 

569 fsm, 

570 event_callback=self._handle_event, 

571 signal_detector=signal_detector, 

572 handoff_handler=handoff_handler, 

573 loops_dir=self.loops_dir, 

574 **executor_kwargs, 

575 ) 

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

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

578 self._continuation_prompt: str | None = None 

579 self.event_bus = EventBus() 

580 

581 @property 

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

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

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

585 

586 @_on_event.setter 

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

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

589 self.event_bus._observers.clear() 

590 if callback is not None: 

591 self.event_bus.register(callback) 

592 

593 def close_transports(self) -> None: 

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

595 self.event_bus.close_transports() 

596 

597 def request_shutdown(self) -> None: 

598 """Request graceful shutdown of the executor. 

599 

600 Delegates to the underlying FSMExecutor's request_shutdown method. 

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

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

603 """ 

604 self._executor.request_shutdown() 

605 

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

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

608 

609 Args: 

610 event: Event dictionary from executor 

611 """ 

612 self.persistence.append_event(event) 

613 

614 # Save state after state transitions 

615 event_type = event.get("event") 

616 if event_type in ("state_enter", "loop_complete"): 

617 self._save_state() 

618 

619 # Track evaluation results for state persistence 

620 if event_type == "evaluate": 

621 self._last_result = { 

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

623 "details": { 

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

625 }, 

626 } 

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

628 if eval_type != "llm_structured": 

629 self._last_non_llm_result = { 

630 "state": self._executor.current_state, 

631 "evaluator": eval_type, 

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

633 "details": { 

634 k: v 

635 for k, v in event.items() 

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

637 }, 

638 } 

639 elif _is_meta_loop(self.fsm): 

640 self._write_meta_eval_entry(event) 

641 

642 # Track handoff events for continuation prompt 

643 if event_type == "handoff_detected": 

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

645 

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

647 self.event_bus.emit(event) 

648 

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

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

651 non_llm = self._last_non_llm_result or {} 

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

653 

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

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

656 agreed: bool | None = ( 

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

658 ) 

659 

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

661 ext_target = non_llm_details.get("target") 

662 

663 entry: dict[str, Any] = { 

664 "iteration": self._executor.iteration, 

665 "ts": _iso_now(), 

666 "loop": self.fsm.name, 

667 "state": self._executor.current_state, 

668 "llm_verdict": llm_verdict, 

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

670 "external_verdict": ext_verdict or None, 

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

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

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

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

675 "diff_stats": _get_diff_stats(), 

676 "agreed": agreed, 

677 } 

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

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

680 

681 def _save_state(self) -> None: 

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

683 status = "running" 

684 if self._executor.current_state: 

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

686 if state_config and state_config.terminal: 

687 status = "completed" 

688 

689 state = LoopState( 

690 loop_name=self.fsm.name, 

691 current_state=self._executor.current_state, 

692 iteration=self._executor.iteration, 

693 captured=self._executor.captured, 

694 prev_result=self._executor.prev_result, 

695 last_result=self._last_result, 

696 started_at=self._executor.started_at, 

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

698 status=status, 

699 accumulated_ms=_now_ms() 

700 - self._executor.start_time_ms 

701 + self._executor.elapsed_offset_ms, 

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

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

704 consecutive_rate_limit_exhaustions=(self._executor._consecutive_rate_limit_exhaustions), 

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

706 pid=self._run_pid, 

707 ) 

708 self.persistence.save_state(state) 

709 

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

711 """Run the FSM with persistence. 

712 

713 Args: 

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

715 

716 Returns: 

717 ExecutionResult from the execution 

718 """ 

719 if clear_previous: 

720 self.persistence.clear_all() 

721 

722 result = self._executor.run() 

723 

724 # Update final state 

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

726 if result.terminated_by in ("max_iterations", "signal"): 

727 final_status = "interrupted" 

728 if result.terminated_by == "handoff": 

729 final_status = "awaiting_continuation" 

730 if result.terminated_by == "timeout": 

731 final_status = "timed_out" 

732 if result.terminated_by == "cycle_detected": 

733 final_status = "failed" 

734 

735 final_state = LoopState( 

736 loop_name=self.fsm.name, 

737 current_state=result.final_state, 

738 iteration=result.iterations, 

739 captured=result.captured, 

740 prev_result=self._executor.prev_result, 

741 last_result=self._last_result, 

742 started_at=self._executor.started_at, 

743 updated_at="", 

744 status=final_status, 

745 continuation_prompt=self._continuation_prompt, 

746 accumulated_ms=result.duration_ms, 

747 ) 

748 self.persistence.save_state(final_state) 

749 self.persistence.archive_run() 

750 

751 return result 

752 

753 def resume(self) -> ExecutionResult | None: 

754 """Resume from saved state, or None if no resumable state. 

755 

756 Resumable states are: "running", "awaiting_continuation", and "interrupted". 

757 

758 Returns: 

759 ExecutionResult if resumed and completed, None if no resumable state 

760 """ 

761 state = self.persistence.load_state() 

762 if state is None: 

763 return None 

764 

765 if state.status not in RESUMABLE_STATUSES: 

766 return None # Already completed/failed 

767 

768 # Restore executor state 

769 self._executor.current_state = state.current_state 

770 self._executor.iteration = state.iteration 

771 self._executor.captured = state.captured 

772 self._executor.prev_result = state.prev_result 

773 self._executor.started_at = state.started_at 

774 self._last_result = state.last_result 

775 self._executor._retry_counts = dict(state.retry_counts) 

776 self._executor._rate_limit_retries = { 

777 k: dict(v) for k, v in state.rate_limit_retries.items() 

778 } 

779 self._executor._consecutive_rate_limit_exhaustions = ( 

780 state.consecutive_rate_limit_exhaustions 

781 ) 

782 self._executor._edge_revisit_counts = dict(state.edge_revisit_counts) 

783 

784 # Restore accumulated elapsed time so duration_ms and ${loop.elapsed_ms} reflect 

785 # the full loop lifetime (all segments), not just the resumed segment. 

786 # FSMExecutor.run() will reset start_time_ms to _now_ms(), so we use elapsed_offset_ms 

787 # to carry forward the time already spent before this resume. 

788 self._executor.elapsed_offset_ms = state.accumulated_ms 

789 

790 # Clear any pending signals from previous run 

791 self._executor._pending_handoff = None 

792 self._executor._pending_error = None 

793 

794 # Emit resume event with continuation context if available 

795 resume_event: dict[str, Any] = { 

796 "event": "loop_resume", 

797 "ts": _iso_now(), 

798 "loop": self.fsm.name, 

799 "from_state": state.current_state, 

800 "iteration": state.iteration, 

801 } 

802 if state.status == "awaiting_continuation" and state.continuation_prompt: 

803 resume_event["from_handoff"] = True 

804 resume_event["continuation_prompt"] = state.continuation_prompt 

805 self.persistence.append_event(resume_event) 

806 self.event_bus.emit(resume_event) 

807 

808 # Continue execution (don't clear previous events) 

809 return self.run(clear_previous=False) 

810 

811 

812def _find_instances(loop_name: str, running_dir: Path) -> list[tuple[str | None, LoopState]]: 

813 """Discover all state-file instances for *loop_name* in *running_dir*. 

814 

815 Globs ``{loop_name}-*.state.json`` for instance-scoped files and 

816 ``{loop_name}.state.json`` for legacy bare-name files. 

817 

818 Returns: 

819 List of ``(instance_id, LoopState)`` tuples sorted by file name. 

820 *instance_id* is the file stem (e.g. ``"autodev-20260503T122306"``) 

821 for instance-scoped files, or ``None`` for legacy bare-name files. 

822 """ 

823 if not running_dir.exists():