Coverage for little_loops / fsm / persistence.py: 45%
464 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-16 13:12 -0500
1"""State persistence and event streaming for FSM loops.
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
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"""
21from __future__ import annotations
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
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
42RUNNING_DIR = ".running"
43HISTORY_DIR = ".history"
45RESUMABLE_STATUSES: frozenset[str] = frozenset({"running", "awaiting_continuation", "interrupted"})
47logger = logging.getLogger(__name__)
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}$")
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
59def _iso_now() -> str:
60 """Return current time as ISO 8601 string."""
61 return datetime.now(UTC).isoformat()
64def _now_ms() -> int:
65 """Return current time in milliseconds."""
66 return int(time.time() * 1000)
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")
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 }
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
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
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.
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
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.
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
162@dataclass
163class LoopState:
164 """Persistent state for an FSM loop execution.
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
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 """
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 messages: list[str] = field(default_factory=list)
217 def to_dict(self) -> dict[str, Any]:
218 """Convert to dictionary for JSON serialization."""
219 result = {
220 "loop_name": self.loop_name,
221 "current_state": self.current_state,
222 "iteration": self.iteration,
223 "captured": self.captured,
224 "prev_result": self.prev_result,
225 "last_result": self.last_result,
226 "started_at": self.started_at,
227 "updated_at": self.updated_at,
228 "status": self.status,
229 "accumulated_ms": self.accumulated_ms,
230 }
231 if self.continuation_prompt is not None:
232 result["continuation_prompt"] = self.continuation_prompt
233 if self.retry_counts:
234 result["retry_counts"] = self.retry_counts
235 if self.rate_limit_retries:
236 result["rate_limit_retries"] = self.rate_limit_retries
237 if self.consecutive_rate_limit_exhaustions:
238 result["consecutive_rate_limit_exhaustions"] = self.consecutive_rate_limit_exhaustions
239 if self.edge_revisit_counts:
240 result["edge_revisit_counts"] = self.edge_revisit_counts
241 if self.active_sub_loop is not None:
242 result["active_sub_loop"] = self.active_sub_loop
243 if self.pid is not None:
244 result["pid"] = self.pid
245 if self.reconciled_at is not None:
246 result["reconciled_at"] = self.reconciled_at
247 if self.messages:
248 result["messages"] = self.messages
249 return result
251 @classmethod
252 def from_dict(cls, data: dict[str, Any]) -> LoopState:
253 """Create LoopState from dictionary.
255 Migrates legacy ``rate_limit_retries`` values from ``dict[str, int]``
256 (BUG-1107 pre-ENH-1133 shape) to the dict-of-record shape. Integer
257 values are coerced to ``{"short_retries": <int>, "long_retries": 0,
258 "total_wait_seconds": 0.0, "first_seen_at": None}``.
260 Args:
261 data: Dictionary with loop state fields
263 Returns:
264 LoopState instance
265 """
266 raw_rl = data.get("rate_limit_retries", {}) or {}
267 migrated_rl: dict[str, dict[str, Any]] = {}
268 for state_name, value in raw_rl.items():
269 if isinstance(value, int):
270 migrated_rl[state_name] = {
271 "short_retries": value,
272 "long_retries": 0,
273 "total_wait_seconds": 0.0,
274 "first_seen_at": None,
275 }
276 elif isinstance(value, dict):
277 migrated_rl[state_name] = value
278 return cls(
279 loop_name=data["loop_name"],
280 current_state=data["current_state"],
281 iteration=data["iteration"],
282 captured=data.get("captured", {}),
283 prev_result=data.get("prev_result"),
284 last_result=data.get("last_result"),
285 started_at=data["started_at"],
286 updated_at=data.get("updated_at", ""),
287 status=data["status"],
288 continuation_prompt=data.get("continuation_prompt"),
289 accumulated_ms=data.get("accumulated_ms", 0),
290 retry_counts=data.get("retry_counts", {}),
291 rate_limit_retries=migrated_rl,
292 consecutive_rate_limit_exhaustions=data.get("consecutive_rate_limit_exhaustions", 0),
293 edge_revisit_counts=data.get("edge_revisit_counts", {}),
294 active_sub_loop=data.get("active_sub_loop"),
295 pid=data.get("pid"),
296 reconciled_at=data.get("reconciled_at"),
297 messages=data.get("messages", []),
298 )
301class StatePersistence:
302 """Manage loop state persistence and event streaming.
304 Handles file I/O for:
305 - State file: JSON file with current execution state
306 - Events file: JSONL file with execution events (append-only)
308 Files are stored in .loops/.running/<instance_id>.*
309 """
311 def __init__(
312 self, loop_name: str, loops_dir: Path | None = None, instance_id: str | None = None
313 ) -> None:
314 """Initialize persistence for a loop.
316 Args:
317 loop_name: Name of the loop
318 loops_dir: Base directory for loops (default: .loops)
319 instance_id: Optional unique instance identifier; falls back to loop_name when None
320 """
321 self.loop_name = loop_name
322 self.loops_dir = loops_dir or Path(".loops")
323 self.running_dir = self.loops_dir / RUNNING_DIR
324 stem = instance_id or loop_name
325 self.state_file = self.running_dir / f"{stem}.state.json"
326 self.events_file = self.running_dir / f"{stem}.events.jsonl"
327 self.meta_eval_file = self.running_dir / f"{stem}.meta-eval.jsonl"
329 def initialize(self) -> None:
330 """Create running directory if needed."""
331 self.running_dir.mkdir(parents=True, exist_ok=True)
333 def save_state(self, state: LoopState) -> None:
334 """Save current state to file using an atomic write.
336 Updates the updated_at timestamp before saving. Writes to a temporary
337 file first, then renames it over the target to avoid leaving a corrupt
338 or empty state file if the process is killed mid-write.
340 Args:
341 state: LoopState to save
342 """
343 state.updated_at = _iso_now()
344 data = json.dumps(state.to_dict(), indent=2)
345 tmp_fd, tmp_path = tempfile.mkstemp(dir=self.state_file.parent, suffix=".tmp")
346 try:
347 with os.fdopen(tmp_fd, "w") as f:
348 f.write(data)
349 os.replace(tmp_path, self.state_file)
350 except Exception:
351 os.unlink(tmp_path)
352 raise
354 def load_state(self) -> LoopState | None:
355 """Load state from file, or None if not exists.
357 Returns:
358 LoopState if file exists and is valid, None otherwise
359 """
360 if not self.state_file.exists():
361 return None
362 try:
363 data = json.loads(self.state_file.read_text())
364 except json.JSONDecodeError:
365 return None
366 try:
367 return LoopState.from_dict(data)
368 except KeyError as e:
369 logger.warning("Corrupted state file %s: missing key %s", self.state_file, e)
370 return None
372 def clear_state(self) -> None:
373 """Remove state file."""
374 if self.state_file.exists():
375 self.state_file.unlink()
377 def append_event(self, event: dict[str, Any]) -> None:
378 """Append event to JSONL file.
380 Args:
381 event: Event dictionary to append
382 """
383 with open(self.events_file, "a", encoding="utf-8") as f:
384 f.write(json.dumps(event) + "\n")
386 def read_events(self) -> list[dict[str, Any]]:
387 """Read all events from file.
389 Returns:
390 List of event dictionaries, empty if file doesn't exist
391 """
392 if not self.events_file.exists():
393 return []
394 events: list[dict[str, Any]] = []
395 with open(self.events_file, encoding="utf-8") as f:
396 for line in f:
397 line = line.strip()
398 if line:
399 try:
400 events.append(json.loads(line))
401 except json.JSONDecodeError:
402 continue # Skip malformed lines
403 return events
405 def clear_events(self) -> None:
406 """Remove events file."""
407 if self.events_file.exists():
408 self.events_file.unlink()
410 def clear_meta_eval(self) -> None:
411 """Remove meta-eval file."""
412 if self.meta_eval_file.exists():
413 self.meta_eval_file.unlink()
415 def archive_run(self) -> Path | None:
416 """Archive current run files to .history/ before clearing.
418 Reads the current state to derive the run timestamp, then copies
419 both state.json and events.jsonl into:
420 <loops_dir>/.history/<run_id>-<loop_name>/
422 where run_id is a compact ISO timestamp derived from started_at
423 (e.g. "2024-01-15T103000" from "2024-01-15T10:30:00.123456+00:00").
425 Returns:
426 Path to the archive directory if files were archived, None if
427 there were no files to archive (fresh run).
428 """
429 has_state = self.state_file.exists()
430 has_events = self.events_file.exists()
431 if not has_state and not has_events:
432 return None
434 # Derive run ID from started_at in state file, or fall back to now
435 state = self.load_state()
436 if state is not None and state.started_at:
437 # Compact ISO: strip colons, dots, plus signs; take first 19 chars
438 # e.g. "2024-01-15T10:30:00.123+00:00" → "2024-01-15T103000"
439 run_id = state.started_at.replace(":", "").replace(".", "").replace("+", "")[:17]
440 else:
441 run_id = datetime.now(UTC).strftime("%Y-%m-%dT%H%M%S")
443 run_folder = f"{run_id}-{self.loop_name}"
444 archive_dir = self.loops_dir / HISTORY_DIR / run_folder
445 archive_dir.mkdir(parents=True, exist_ok=True)
447 if has_state:
448 shutil.copy2(self.state_file, archive_dir / "state.json")
449 if has_events:
450 shutil.copy2(self.events_file, archive_dir / "events.jsonl")
451 if self.meta_eval_file.exists():
452 shutil.copy2(self.meta_eval_file, archive_dir / "meta-eval.jsonl")
454 return archive_dir
456 def clear_all(self) -> None:
457 """Archive current run files then clear state and events (for new run)."""
458 self.archive_run()
459 self.clear_state()
460 self.clear_events()
461 self.clear_meta_eval()
464def _reconcile_stale_runs(loops_dir: Path) -> int:
465 """Archive state files in .running/ that belong to dead or terminal processes.
467 Called at loop startup to clean up files left by crashed or interrupted runs.
468 Returns the count of archived files.
470 Strategy (mirrors LockManager.find_conflict() stale-lock cleanup):
471 - Terminal-status files (completed/failed/timed_out) are archived
472 unconditionally — they are definitionally stale by invariant.
473 - status="interrupted" files are left alone so the user can resume them.
474 - status="running" files are checked via their sibling .pid file; archived
475 only if the PID is confirmed dead. No .pid file → leave alone (can't confirm).
476 """
477 running_dir = loops_dir / RUNNING_DIR
478 if not running_dir.exists():
479 return 0
481 terminal_statuses = {"completed", "failed", "timed_out"}
482 archived = 0
484 for state_file in running_dir.glob("*.state.json"):
485 try:
486 data = json.loads(state_file.read_text())
487 state = LoopState.from_dict(data)
488 except (json.JSONDecodeError, KeyError, OSError):
489 continue
491 is_stale = state.status in terminal_statuses
493 if not is_stale and state.status == "running":
494 stem = state_file.name.removesuffix(".state.json")
495 pid_file = running_dir / f"{stem}.pid"
496 if pid_file.exists():
497 try:
498 pid = int(pid_file.read_text().strip())
499 is_stale = not _process_alive(pid)
500 except (OSError, ValueError):
501 pass
503 if not is_stale:
504 continue
506 stem = state_file.name.removesuffix(".state.json")
507 instance_id = stem if stem != state.loop_name else None
508 persistence = StatePersistence(
509 loop_name=state.loop_name,
510 loops_dir=loops_dir,
511 instance_id=instance_id,
512 )
513 try:
514 persistence.clear_all()
515 (running_dir / f"{stem}.pid").unlink(missing_ok=True)
516 archived += 1
517 logger.debug("Archived stale run: %s (status=%s)", stem, state.status)
518 except OSError as e:
519 logger.warning("Failed to archive stale run %s: %s", stem, e)
521 if archived:
522 logger.info("Reconciliation sweep archived %d stale run(s) from .running/", archived)
524 return archived
527class PersistentExecutor:
528 """FSM Executor with state persistence and event streaming.
530 Wraps FSMExecutor to:
531 - Save state after each state transition
532 - Append events to JSONL file as they occur
533 - Support resuming from saved state
534 - Support graceful shutdown via signal handling
535 """
537 def __init__(
538 self,
539 fsm: FSMLoop,
540 persistence: StatePersistence | None = None,
541 loops_dir: Path | None = None,
542 instance_id: str | None = None,
543 pid: int | None = None,
544 **executor_kwargs: Any,
545 ) -> None:
546 """Initialize persistent executor.
548 Args:
549 fsm: FSM loop definition
550 persistence: Optional pre-configured persistence (for testing)
551 loops_dir: Base directory for loops (default: .loops)
552 instance_id: Optional unique instance identifier for file path scoping
553 pid: OS PID of the running process; stored in saved state for reconciliation
554 **executor_kwargs: Additional kwargs for FSMExecutor
555 """
556 from little_loops.fsm.handoff_handler import HandoffBehavior, HandoffHandler
557 from little_loops.fsm.signal_detector import SignalDetector
559 self.fsm = fsm
560 self.loops_dir = loops_dir
561 self._run_pid = pid
562 self.persistence = persistence or StatePersistence(
563 fsm.name, loops_dir or Path(".loops"), instance_id=instance_id
564 )
565 self.persistence.initialize()
567 # Create signal detector and handler based on FSM config
568 signal_detector = SignalDetector()
569 handoff_handler = HandoffHandler(HandoffBehavior(fsm.on_handoff))
571 # Create base executor with event callback that persists
572 self._executor = FSMExecutor(
573 fsm,
574 event_callback=self._handle_event,
575 signal_detector=signal_detector,
576 handoff_handler=handoff_handler,
577 loops_dir=self.loops_dir,
578 **executor_kwargs,
579 )
580 self._last_result: dict[str, Any] | None = None
581 self._last_non_llm_result: dict[str, Any] | None = None
582 self._continuation_prompt: str | None = None
583 self.event_bus = EventBus()
585 @property
586 def _on_event(self) -> EventCallback | None:
587 """Backward-compatible access to the first observer on the event bus."""
588 return self.event_bus._observers[0][0] if self.event_bus._observers else None
590 @_on_event.setter
591 def _on_event(self, callback: EventCallback | None) -> None:
592 """Backward-compatible setter: replaces all observers with this one."""
593 self.event_bus._observers.clear()
594 if callback is not None:
595 self.event_bus.register(callback)
597 def close_transports(self) -> None:
598 """Close all transports registered on the underlying EventBus."""
599 self.event_bus.close_transports()
601 def request_shutdown(self) -> None:
602 """Request graceful shutdown of the executor.
604 Delegates to the underlying FSMExecutor's request_shutdown method.
605 The loop will exit cleanly after the current state completes,
606 saving state as "interrupted" so it can be resumed later.
607 """
608 self._executor.request_shutdown()
610 def _handle_event(self, event: dict[str, Any]) -> None:
611 """Handle event: persist to file and save state.
613 Args:
614 event: Event dictionary from executor
615 """
616 self.persistence.append_event(event)
618 event_type = event.get("event")
620 # Write per-state token usage to usage.jsonl when an LLM action completes.
621 # Shell and mcp_tool invocations produce no token data and are skipped.
622 if event_type == "action_complete" and "input_tokens" in event:
623 run_dir = self.fsm.context.get("run_dir", "")
624 if run_dir:
625 usage_path = Path(run_dir) / "usage.jsonl"
626 entry = {
627 "iteration": self._executor.iteration,
628 "state": self._executor.current_state,
629 "action_type": "prompt" if event.get("is_prompt") else "shell_or_mcp",
630 "input_tokens": event["input_tokens"],
631 "output_tokens": event["output_tokens"],
632 "cache_read_tokens": event.get("cache_read_tokens", 0),
633 "cache_creation_tokens": event.get("cache_creation_tokens", 0),
634 "model": event.get("model", "unknown"),
635 "timestamp": event.get("ts", ""),
636 }
637 with open(usage_path, "a", encoding="utf-8") as f:
638 f.write(json.dumps(entry) + "\n")
640 # Append shared message to messages.jsonl when a state appends to the log.
641 if event_type == "messages_append":
642 run_dir = self.fsm.context.get("run_dir", "")
643 if run_dir:
644 messages_path = Path(run_dir) / "messages.jsonl"
645 entry = {
646 "iteration": self._executor.iteration,
647 "state": event.get("state", ""),
648 "message": event.get("message", ""),
649 "timestamp": event.get("ts", ""),
650 }
651 with open(messages_path, "a", encoding="utf-8") as f:
652 f.write(json.dumps(entry) + "\n")
654 # Save state after state transitions
655 if event_type in ("state_enter", "loop_complete", "baseline_complete"):
656 self._save_state()
658 # Track evaluation results for state persistence
659 if event_type == "evaluate":
660 self._last_result = {
661 "verdict": event.get("verdict"),
662 "details": {
663 k: v for k, v in event.items() if k not in ("event", "ts", "type", "verdict")
664 },
665 }
666 eval_type = event.get("type", "")
667 if eval_type != "llm_structured":
668 self._last_non_llm_result = {
669 "state": self._executor.current_state,
670 "evaluator": eval_type,
671 "verdict": event.get("verdict", ""),
672 "details": {
673 k: v
674 for k, v in event.items()
675 if k not in ("event", "ts", "type", "verdict")
676 },
677 }
678 elif _is_meta_loop(self.fsm):
679 self._write_meta_eval_entry(event)
681 # Track handoff events for continuation prompt
682 if event_type == "handoff_detected":
683 self._continuation_prompt = event.get("continuation")
685 # Delegate to registered observers (e.g. progress display, extensions)
686 self.event_bus.emit(event)
688 def _write_meta_eval_entry(self, event: dict[str, Any]) -> None:
689 """Append one JSONL entry to meta-eval.jsonl for an llm_structured evaluate in a meta-loop."""
690 non_llm = self._last_non_llm_result or {}
691 non_llm_details = non_llm.get("details", {})
693 llm_verdict = event.get("verdict", "")
694 ext_verdict = non_llm.get("verdict", "")
695 agreed: bool | None = (
696 _verdict_is_yes(llm_verdict) == _verdict_is_yes(ext_verdict) if ext_verdict else None
697 )
699 ext_value = non_llm_details.get("current", non_llm_details.get("value"))
700 ext_target = non_llm_details.get("target")
702 entry: dict[str, Any] = {
703 "iteration": self._executor.iteration,
704 "ts": _iso_now(),
705 "loop": self.fsm.name,
706 "state": self._executor.current_state,
707 "llm_verdict": llm_verdict,
708 "llm_rationale": (event.get("reason") or "")[:200],
709 "external_verdict": ext_verdict or None,
710 "external_state": non_llm.get("state"),
711 "external_evaluator": non_llm.get("evaluator") or None,
712 "external_value": str(ext_value) if ext_value is not None else None,
713 "external_target": str(ext_target) if ext_target is not None else None,
714 "diff_stats": _get_diff_stats(),
715 "agreed": agreed,
716 }
717 with open(self.persistence.meta_eval_file, "a", encoding="utf-8") as f:
718 f.write(json.dumps(entry) + "\n")
720 def _save_state(self) -> None:
721 """Save current executor state to file."""
722 status = "running"
723 if self._executor.current_state:
724 state_config = self.fsm.states.get(self._executor.current_state)
725 if state_config and state_config.terminal:
726 status = "completed"
728 state = LoopState(
729 loop_name=self.fsm.name,
730 current_state=self._executor.current_state,
731 iteration=self._executor.iteration,
732 captured=self._executor.captured,
733 prev_result=self._executor.prev_result,
734 last_result=self._last_result,
735 started_at=self._executor.started_at,
736 updated_at="", # Will be set by save_state
737 status=status,
738 accumulated_ms=_now_ms()
739 - self._executor.start_time_ms
740 + self._executor.elapsed_offset_ms,
741 retry_counts=dict(self._executor._retry_counts),
742 rate_limit_retries={k: dict(v) for k, v in self._executor._rate_limit_retries.items()},
743 consecutive_rate_limit_exhaustions=(self._executor._consecutive_rate_limit_exhaustions),
744 edge_revisit_counts=dict(self._executor._edge_revisit_counts),
745 pid=self._run_pid,
746 messages=list(self._executor.messages),
747 )
748 self.persistence.save_state(state)
750 def run(self, clear_previous: bool = True) -> ExecutionResult:
751 """Run the FSM with persistence.
753 Args:
754 clear_previous: If True, clear previous state/events before running
756 Returns:
757 ExecutionResult from the execution
758 """
759 if clear_previous:
760 self.persistence.clear_all()
762 result = self._executor.run()
764 # Update final state
765 final_status = "completed" if result.terminated_by == "terminal" else "failed"
766 if result.terminated_by in ("max_iterations", "signal"):
767 final_status = "interrupted"
768 if result.terminated_by == "handoff":
769 final_status = "awaiting_continuation"
770 if result.terminated_by == "timeout":
771 final_status = "timed_out"
772 if result.terminated_by == "cycle_detected":
773 final_status = "failed"
775 final_state = LoopState(
776 loop_name=self.fsm.name,
777 current_state=result.final_state,
778 iteration=result.iterations,
779 captured=result.captured,
780 prev_result=self._executor.prev_result,
781 last_result=self._last_result,
782 started_at=self._executor.started_at,
783 updated_at="",
784 status=final_status,
785 continuation_prompt=self._continuation_prompt,
786 accumulated_ms=result.duration_ms,
787 )
788 self.persistence.save_state(final_state)
789 self.persistence.archive_run()
791 return result
793 def resume(self) -> ExecutionResult | None:
794 """Resume from saved state, or None if no resumable state.
796 Resumable states are: "running", "awaiting_continuation", and "interrupted".
798 Returns:
799 ExecutionResult if resumed and completed, None if no resumable state
800 """
801 state = self.persistence.load_state()
802 if state is None:
803 return None
805 if state.status not in RESUMABLE_STATUSES:
806 return None # Already completed/failed
808 # Restore executor state
809 self._executor.current_state = state.current_state
810 self._executor.iteration = state.iteration
811 self._executor.captured = state.captured
812 self._executor.prev_result = state.prev_result
813 self._executor.started_at = state.started_at
814 self._last_result = state.last_result
815 self._executor._retry_counts = dict(state.retry_counts)
816 self._executor._rate_limit_retries = {
817 k: dict(v) for k, v in state.rate_limit_retries.items()
818 }
819 self._executor._consecutive_rate_limit_exhaustions = (
820 state.consecutive_rate_limit_exhaustions
821 )
822 self._executor._edge_revisit_counts = dict(state.edge_revisit_counts)
823 self._executor.messages = list(state.messages)
825 # Restore accumulated elapsed time so duration_ms and ${loop.elapsed_ms} reflect
826 # the full loop lifetime (all segments), not just the resumed segment.
827 # FSMExecutor.run() will reset start_time_ms to _now_ms(), so we use elapsed_offset_ms
828 # to carry forward the time already spent before this resume.
829 self._executor.elapsed_offset_ms = state.accumulated_ms
831 # Clear any pending signals from previous run
832 self._executor._pending_handoff = None
833 self._executor._pending_error = None
835 # Emit resume event with continuation context if available
836 resume_event: dict[str, Any] = {
837 "event": "loop_resume",
838 "ts": _iso_now(),
839 "loop": self.fsm.name,
840 "from_state": state.current_state,
841 "iteration": state.iteration,
842 }
843 if state.status == "awaiting_continuation" and state.continuation_prompt:
844 resume_event["from_handoff"] = True
845 resume_event["continuation_prompt"] = state.continuation_prompt
846 self.persistence.append_event(resume_event)
847 self.event_bus.emit(resume_event)
849 # Continue execution (don't clear previous events)
850 return self.run(clear_previous=False)
853def _find_instances(loop_name: str, running_dir: Path) -> list[tuple[str | None, LoopState]]:
854 """Discover all state-file instances for *loop_name* in *running_dir*.
856 Globs ``{loop_name}-*.state.json`` for instance-scoped files and
857 ``{loop_name}.state.json`` for legacy bare-name files.
859 Returns:
860 List of ``(instance_id, LoopState)`` tuples sorted by file name.
861 *instance_id* is the file stem (e.g. ``"autodev-20260503T122306"``)
862 for instance-scoped files, or ``None`` for legacy bare-name files.
863 """
864 if not running_dir.exists():
865 return []
867 instances: list[tuple[str | None, LoopState]] = []
869 # Instance-scoped files: {loop_name}-YYYYMMDDTHHMMSS.state.json
870 # Use Path(stem).stem to strip both suffixes (.state.json → base stem).
871 for state_file in sorted(running_dir.glob(f"{loop_name}-*.state.json")):
872 base_stem = Path(state_file.stem).stem # e.g. "autodev-20260503T122306"
873 if not _INSTANCE_SUFFIX.search(base_stem):
874 continue # skip files like "loop-name-extra" that don't match timestamp pattern
875 try:
876 data = json.loads(state_file.read_text())
877 instances.append((base_stem, LoopState.from_dict(data)))
878 except (json.JSONDecodeError, KeyError):
879 continue
881 # Legacy bare-name file: {loop_name}.state.json
882 legacy_file = running_dir / f"{loop_name}.state.json"
883 if legacy_file.exists():
884 try:
885 data = json.loads(legacy_file.read_text())
886 instances.append((None, LoopState.from_dict(data)))
887 except (json.JSONDecodeError, KeyError):
888 pass
890 return instances
893def list_running_loops(loops_dir: Path | None = None) -> list[LoopState]:
894 """List all loops with saved state.
896 Args:
897 loops_dir: Base directory for loops (default: .loops)
899 Returns:
900 List of LoopState objects for all loops with state files
901 """
902 base_dir = loops_dir or Path(".loops")
903 running_dir = base_dir / RUNNING_DIR
905 if not running_dir.exists():
906 return []
908 states: list[LoopState] = []
909 for state_file in running_dir.glob("*.state.json"):
910 try:
911 data = json.loads(state_file.read_text())
912 state = LoopState.from_dict(data)
913 except (json.JSONDecodeError, KeyError):
914 continue # Skip malformed files
915 stem = state_file.stem.removesuffix(".state")
916 persistence = StatePersistence(
917 state.loop_name, base_dir, instance_id=stem if stem != state.loop_name else None
918 )
919 state = _reconcile_stale_running(state, persistence, running_dir, stem)
920 states.append(state)
922 # Include loops that have a PID file but no state file yet (still starting up).
923 # Strip instance-ID timestamp suffix (e.g. "autodev-20240115T103000" → "autodev")
924 # before the known_names check to avoid spurious "starting" entries for loops
925 # that already have a state file under their logical name.
926 known_names = {s.loop_name for s in states}
927 for pid_file in running_dir.glob("*.pid"):
928 logical_name = _INSTANCE_SUFFIX.sub("", pid_file.stem)
929 if logical_name in known_names:
930 continue # state file already covers this loop
931 try:
932 pid = int(pid_file.read_text().strip())
933 except (ValueError, OSError):
934 continue
935 if _process_alive(pid):
936 states.append(
937 LoopState(
938 loop_name=logical_name,
939 current_state="(initializing)",
940 iteration=0,
941 captured={},
942 prev_result=None,
943 last_result=None,
944 started_at="",
945 updated_at="",
946 status="starting",
947 )
948 )
950 return states
953def list_run_history(loop_name: str, loops_dir: Path | None = None) -> list[LoopState]:
954 """List archived runs for a loop, newest first.
956 Reads state files from .loops/.history/<run_id>-<loop_name>/state.json and
957 returns them sorted by started_at descending (most recent run first).
959 Also checks the legacy nested layout .loops/.history/<loop_name>/*/state.json
960 for backward compatibility with existing history folders.
962 Args:
963 loop_name: Name of the loop
964 loops_dir: Base directory for loops (default: .loops)
966 Returns:
967 List of LoopState objects for all archived runs, newest first.
968 Returns an empty list if no history exists.
969 """
970 base_dir = loops_dir or Path(".loops")
971 history_dir = base_dir / HISTORY_DIR
973 if not history_dir.exists():
974 return []
976 states: list[LoopState] = []
978 # Flat layout: <run_id>-<loop_name>/state.json
979 for state_file in history_dir.glob(f"*-{loop_name}/state.json"):
980 try:
981 data = json.loads(state_file.read_text())
982 states.append(LoopState.from_dict(data))
983 except (json.JSONDecodeError, KeyError):
984 continue
986 # Backward compat: legacy nested layout <loop_name>/<run_id>/state.json
987 old_loop_dir = history_dir / loop_name
988 if old_loop_dir.exists():
989 logger.warning(
990 "Found legacy nested history at %s; migrate to flat layout by moving "
991 "each run to .history/<run_id>-%s/",
992 old_loop_dir,
993 loop_name,
994 )
995 for state_file in old_loop_dir.glob("*/state.json"):
996 try:
997 data = json.loads(state_file.read_text())
998 states.append(LoopState.from_dict(data))
999 except (json.JSONDecodeError, KeyError):
1000 continue
1002 states.sort(key=lambda s: s.started_at, reverse=True)
1003 return states
1006def get_archived_events(
1007 loop_name: str, run_id: str, loops_dir: Path | None = None
1008) -> list[dict[str, Any]]:
1009 """Read events for a specific archived run.
1011 Args:
1012 loop_name: Name of the loop
1013 run_id: The run directory name (compact timestamp)
1014 loops_dir: Base directory for loops (default: .loops)
1016 Returns:
1017 List of event dictionaries, empty if not found.
1018 """
1019 base_dir = loops_dir or Path(".loops")
1020 run_folder = f"{run_id}-{loop_name}"
1021 events_file = base_dir / HISTORY_DIR / run_folder / "events.jsonl"
1023 if not events_file.exists():
1024 return []
1026 events: list[dict[str, Any]] = []
1027 with open(events_file, encoding="utf-8") as f:
1028 for line in f:
1029 line = line.strip()
1030 if line:
1031 try:
1032 events.append(json.loads(line))
1033 except json.JSONDecodeError:
1034 continue
1035 return events
1038def get_loop_history(loop_name: str, loops_dir: Path | None = None) -> list[dict[str, Any]]:
1039 """Get event history for a loop.
1041 Args:
1042 loop_name: Name of the loop
1043 loops_dir: Base directory for loops (default: .loops)
1045 Returns:
1046 List of event dictionaries
1047 """
1048 persistence = StatePersistence(loop_name, loops_dir)
1049 return persistence.read_events()