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