Coverage for little_loops / fsm / persistence.py: 38%
272 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:20 -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.state.json
14 │ └── fix-types.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 tempfile
29import time
30from dataclasses import dataclass, field
31from datetime import UTC, datetime
32from pathlib import Path
33from typing import Any
35from little_loops.events import EventBus
36from little_loops.fsm.concurrency import _process_alive
37from little_loops.fsm.executor import EventCallback, ExecutionResult, FSMExecutor
38from little_loops.fsm.schema import FSMLoop
40RUNNING_DIR = ".running"
41HISTORY_DIR = ".history"
43logger = logging.getLogger(__name__)
45_RUN_FOLDER = re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{6})-(.+)$")
48def _parse_run_folder(name: str) -> tuple[str, str] | None:
49 """Return (run_id, loop_name) from a flat history folder name, or None."""
50 m = _RUN_FOLDER.match(name)
51 return (m.group(1), m.group(2)) if m else None
54def _iso_now() -> str:
55 """Return current time as ISO 8601 string."""
56 return datetime.now(UTC).isoformat()
59def _now_ms() -> int:
60 """Return current time in milliseconds."""
61 return int(time.time() * 1000)
64@dataclass
65class LoopState:
66 """Persistent state for an FSM loop execution.
68 This captures all runtime state needed to resume a loop:
69 - Current state and iteration
70 - Captured variables and previous result
71 - Last evaluation result
72 - Timestamps and status
74 Attributes:
75 loop_name: Name of the loop
76 current_state: Current FSM state name
77 iteration: Current iteration count (1-based)
78 captured: Captured action outputs by variable name
79 prev_result: Previous state's result (output, exit_code, state)
80 last_result: Last evaluation result (verdict, details)
81 started_at: ISO timestamp when loop started
82 updated_at: ISO timestamp when state was last saved
83 status: Execution status (running, completed, failed, interrupted, awaiting_continuation, timed_out)
84 continuation_prompt: Continuation context from handoff signal (if status is awaiting_continuation)
85 accumulated_ms: Total milliseconds elapsed across all segments up to this save (used to restore
86 elapsed time correctly after resume, so duration_ms and ${loop.elapsed_ms} reflect the
87 full loop lifetime rather than only the most recent segment)
88 """
90 loop_name: str
91 current_state: str
92 iteration: int
93 captured: dict[str, dict[str, Any]]
94 prev_result: dict[str, Any] | None
95 last_result: dict[str, Any] | None
96 started_at: str
97 updated_at: str
98 status: (
99 str # "running", "completed", "failed", "interrupted", "awaiting_continuation", "timed_out"
100 )
101 continuation_prompt: str | None = None
102 accumulated_ms: int = 0 # total elapsed ms across all segments (for resume offset)
103 retry_counts: dict[str, int] = field(default_factory=dict) # per-state retry tracking
104 active_sub_loop: str | None = None # name of currently executing sub-loop (observability)
106 def to_dict(self) -> dict[str, Any]:
107 """Convert to dictionary for JSON serialization."""
108 result = {
109 "loop_name": self.loop_name,
110 "current_state": self.current_state,
111 "iteration": self.iteration,
112 "captured": self.captured,
113 "prev_result": self.prev_result,
114 "last_result": self.last_result,
115 "started_at": self.started_at,
116 "updated_at": self.updated_at,
117 "status": self.status,
118 "accumulated_ms": self.accumulated_ms,
119 }
120 if self.continuation_prompt is not None:
121 result["continuation_prompt"] = self.continuation_prompt
122 if self.retry_counts:
123 result["retry_counts"] = self.retry_counts
124 if self.active_sub_loop is not None:
125 result["active_sub_loop"] = self.active_sub_loop
126 return result
128 @classmethod
129 def from_dict(cls, data: dict[str, Any]) -> LoopState:
130 """Create LoopState from dictionary.
132 Args:
133 data: Dictionary with loop state fields
135 Returns:
136 LoopState instance
137 """
138 return cls(
139 loop_name=data["loop_name"],
140 current_state=data["current_state"],
141 iteration=data["iteration"],
142 captured=data.get("captured", {}),
143 prev_result=data.get("prev_result"),
144 last_result=data.get("last_result"),
145 started_at=data["started_at"],
146 updated_at=data.get("updated_at", ""),
147 status=data["status"],
148 continuation_prompt=data.get("continuation_prompt"),
149 accumulated_ms=data.get("accumulated_ms", 0),
150 retry_counts=data.get("retry_counts", {}),
151 active_sub_loop=data.get("active_sub_loop"),
152 )
155class StatePersistence:
156 """Manage loop state persistence and event streaming.
158 Handles file I/O for:
159 - State file: JSON file with current execution state
160 - Events file: JSONL file with execution events (append-only)
162 Files are stored in .loops/.running/<loop_name>.*
163 """
165 def __init__(self, loop_name: str, loops_dir: Path | None = None) -> None:
166 """Initialize persistence for a loop.
168 Args:
169 loop_name: Name of the loop
170 loops_dir: Base directory for loops (default: .loops)
171 """
172 self.loop_name = loop_name
173 self.loops_dir = loops_dir or Path(".loops")
174 self.running_dir = self.loops_dir / RUNNING_DIR
175 self.state_file = self.running_dir / f"{loop_name}.state.json"
176 self.events_file = self.running_dir / f"{loop_name}.events.jsonl"
178 def initialize(self) -> None:
179 """Create running directory if needed."""
180 self.running_dir.mkdir(parents=True, exist_ok=True)
182 def save_state(self, state: LoopState) -> None:
183 """Save current state to file using an atomic write.
185 Updates the updated_at timestamp before saving. Writes to a temporary
186 file first, then renames it over the target to avoid leaving a corrupt
187 or empty state file if the process is killed mid-write.
189 Args:
190 state: LoopState to save
191 """
192 state.updated_at = _iso_now()
193 data = json.dumps(state.to_dict(), indent=2)
194 tmp_fd, tmp_path = tempfile.mkstemp(dir=self.state_file.parent, suffix=".tmp")
195 try:
196 with os.fdopen(tmp_fd, "w") as f:
197 f.write(data)
198 os.replace(tmp_path, self.state_file)
199 except Exception:
200 os.unlink(tmp_path)
201 raise
203 def load_state(self) -> LoopState | None:
204 """Load state from file, or None if not exists.
206 Returns:
207 LoopState if file exists and is valid, None otherwise
208 """
209 if not self.state_file.exists():
210 return None
211 try:
212 data = json.loads(self.state_file.read_text())
213 except json.JSONDecodeError:
214 return None
215 try:
216 return LoopState.from_dict(data)
217 except KeyError as e:
218 logger.warning("Corrupted state file %s: missing key %s", self.state_file, e)
219 return None
221 def clear_state(self) -> None:
222 """Remove state file."""
223 if self.state_file.exists():
224 self.state_file.unlink()
226 def append_event(self, event: dict[str, Any]) -> None:
227 """Append event to JSONL file.
229 Args:
230 event: Event dictionary to append
231 """
232 with open(self.events_file, "a", encoding="utf-8") as f:
233 f.write(json.dumps(event) + "\n")
235 def read_events(self) -> list[dict[str, Any]]:
236 """Read all events from file.
238 Returns:
239 List of event dictionaries, empty if file doesn't exist
240 """
241 if not self.events_file.exists():
242 return []
243 events: list[dict[str, Any]] = []
244 with open(self.events_file, encoding="utf-8") as f:
245 for line in f:
246 line = line.strip()
247 if line:
248 try:
249 events.append(json.loads(line))
250 except json.JSONDecodeError:
251 continue # Skip malformed lines
252 return events
254 def clear_events(self) -> None:
255 """Remove events file."""
256 if self.events_file.exists():
257 self.events_file.unlink()
259 def archive_run(self) -> Path | None:
260 """Archive current run files to .history/ before clearing.
262 Reads the current state to derive the run timestamp, then copies
263 both state.json and events.jsonl into:
264 <loops_dir>/.history/<run_id>-<loop_name>/
266 where run_id is a compact ISO timestamp derived from started_at
267 (e.g. "2024-01-15T103000" from "2024-01-15T10:30:00.123456+00:00").
269 Returns:
270 Path to the archive directory if files were archived, None if
271 there were no files to archive (fresh run).
272 """
273 has_state = self.state_file.exists()
274 has_events = self.events_file.exists()
275 if not has_state and not has_events:
276 return None
278 # Derive run ID from started_at in state file, or fall back to now
279 state = self.load_state()
280 if state is not None and state.started_at:
281 # Compact ISO: strip colons, dots, plus signs; take first 19 chars
282 # e.g. "2024-01-15T10:30:00.123+00:00" → "2024-01-15T103000"
283 run_id = state.started_at.replace(":", "").replace(".", "").replace("+", "")[:17]
284 else:
285 run_id = datetime.now(UTC).strftime("%Y-%m-%dT%H%M%S")
287 run_folder = f"{run_id}-{self.loop_name}"
288 archive_dir = self.loops_dir / HISTORY_DIR / run_folder
289 archive_dir.mkdir(parents=True, exist_ok=True)
291 if has_state:
292 shutil.copy2(self.state_file, archive_dir / "state.json")
293 if has_events:
294 shutil.copy2(self.events_file, archive_dir / "events.jsonl")
296 return archive_dir
298 def clear_all(self) -> None:
299 """Archive current run files then clear state and events (for new run)."""
300 self.archive_run()
301 self.clear_state()
302 self.clear_events()
305class PersistentExecutor:
306 """FSM Executor with state persistence and event streaming.
308 Wraps FSMExecutor to:
309 - Save state after each state transition
310 - Append events to JSONL file as they occur
311 - Support resuming from saved state
312 - Support graceful shutdown via signal handling
313 """
315 def __init__(
316 self,
317 fsm: FSMLoop,
318 persistence: StatePersistence | None = None,
319 loops_dir: Path | None = None,
320 **executor_kwargs: Any,
321 ) -> None:
322 """Initialize persistent executor.
324 Args:
325 fsm: FSM loop definition
326 persistence: Optional pre-configured persistence (for testing)
327 loops_dir: Base directory for loops (default: .loops)
328 **executor_kwargs: Additional kwargs for FSMExecutor
329 """
330 from little_loops.fsm.handoff_handler import HandoffBehavior, HandoffHandler
331 from little_loops.fsm.signal_detector import SignalDetector
333 self.fsm = fsm
334 self.loops_dir = loops_dir
335 self.persistence = persistence or StatePersistence(fsm.name, loops_dir or Path(".loops"))
336 self.persistence.initialize()
338 # Create signal detector and handler based on FSM config
339 signal_detector = SignalDetector()
340 handoff_handler = HandoffHandler(HandoffBehavior(fsm.on_handoff))
342 # Create base executor with event callback that persists
343 self._executor = FSMExecutor(
344 fsm,
345 event_callback=self._handle_event,
346 signal_detector=signal_detector,
347 handoff_handler=handoff_handler,
348 loops_dir=self.loops_dir,
349 **executor_kwargs,
350 )
351 self._last_result: dict[str, Any] | None = None
352 self._continuation_prompt: str | None = None
353 self.event_bus = EventBus()
355 @property
356 def _on_event(self) -> EventCallback | None:
357 """Backward-compatible access to the first observer on the event bus."""
358 return self.event_bus._observers[0][0] if self.event_bus._observers else None
360 @_on_event.setter
361 def _on_event(self, callback: EventCallback | None) -> None:
362 """Backward-compatible setter: replaces all observers with this one."""
363 self.event_bus._observers.clear()
364 if callback is not None:
365 self.event_bus.register(callback)
367 def request_shutdown(self) -> None:
368 """Request graceful shutdown of the executor.
370 Delegates to the underlying FSMExecutor's request_shutdown method.
371 The loop will exit cleanly after the current state completes,
372 saving state as "interrupted" so it can be resumed later.
373 """
374 self._executor.request_shutdown()
376 def _handle_event(self, event: dict[str, Any]) -> None:
377 """Handle event: persist to file and save state.
379 Args:
380 event: Event dictionary from executor
381 """
382 self.persistence.append_event(event)
384 # Save state after state transitions
385 event_type = event.get("event")
386 if event_type in ("state_enter", "loop_complete"):
387 self._save_state()
389 # Track evaluation results for state persistence
390 if event_type == "evaluate":
391 self._last_result = {
392 "verdict": event.get("verdict"),
393 "details": {
394 k: v for k, v in event.items() if k not in ("event", "ts", "type", "verdict")
395 },
396 }
398 # Track handoff events for continuation prompt
399 if event_type == "handoff_detected":
400 self._continuation_prompt = event.get("continuation")
402 # Delegate to registered observers (e.g. progress display, extensions)
403 self.event_bus.emit(event)
405 def _save_state(self) -> None:
406 """Save current executor state to file."""
407 status = "running"
408 if self._executor.current_state:
409 state_config = self.fsm.states.get(self._executor.current_state)
410 if state_config and state_config.terminal:
411 status = "completed"
413 state = LoopState(
414 loop_name=self.fsm.name,
415 current_state=self._executor.current_state,
416 iteration=self._executor.iteration,
417 captured=self._executor.captured,
418 prev_result=self._executor.prev_result,
419 last_result=self._last_result,
420 started_at=self._executor.started_at,
421 updated_at="", # Will be set by save_state
422 status=status,
423 accumulated_ms=_now_ms()
424 - self._executor.start_time_ms
425 + self._executor.elapsed_offset_ms,
426 retry_counts=dict(self._executor._retry_counts),
427 )
428 self.persistence.save_state(state)
430 def run(self, clear_previous: bool = True) -> ExecutionResult:
431 """Run the FSM with persistence.
433 Args:
434 clear_previous: If True, clear previous state/events before running
436 Returns:
437 ExecutionResult from the execution
438 """
439 if clear_previous:
440 self.persistence.clear_all()
442 result = self._executor.run()
444 # Update final state
445 final_status = "completed" if result.terminated_by == "terminal" else "failed"
446 if result.terminated_by in ("max_iterations", "signal"):
447 final_status = "interrupted"
448 if result.terminated_by == "handoff":
449 final_status = "awaiting_continuation"
450 if result.terminated_by == "timeout":
451 final_status = "timed_out"
453 final_state = LoopState(
454 loop_name=self.fsm.name,
455 current_state=result.final_state,
456 iteration=result.iterations,
457 captured=result.captured,
458 prev_result=self._executor.prev_result,
459 last_result=self._last_result,
460 started_at=self._executor.started_at,
461 updated_at="",
462 status=final_status,
463 continuation_prompt=self._continuation_prompt,
464 accumulated_ms=result.duration_ms,
465 )
466 self.persistence.save_state(final_state)
468 return result
470 def resume(self) -> ExecutionResult | None:
471 """Resume from saved state, or None if no resumable state.
473 Resumable states are: "running" and "awaiting_continuation".
475 Returns:
476 ExecutionResult if resumed and completed, None if no resumable state
477 """
478 state = self.persistence.load_state()
479 if state is None:
480 return None
482 if state.status not in ("running", "awaiting_continuation"):
483 return None # Already completed/failed
485 # Restore executor state
486 self._executor.current_state = state.current_state
487 self._executor.iteration = state.iteration
488 self._executor.captured = state.captured
489 self._executor.prev_result = state.prev_result
490 self._executor.started_at = state.started_at
491 self._last_result = state.last_result
492 self._executor._retry_counts = dict(state.retry_counts)
494 # Restore accumulated elapsed time so duration_ms and ${loop.elapsed_ms} reflect
495 # the full loop lifetime (all segments), not just the resumed segment.
496 # FSMExecutor.run() will reset start_time_ms to _now_ms(), so we use elapsed_offset_ms
497 # to carry forward the time already spent before this resume.
498 self._executor.elapsed_offset_ms = state.accumulated_ms
500 # Clear any pending signals from previous run
501 self._executor._pending_handoff = None
502 self._executor._pending_error = None
504 # Emit resume event with continuation context if available
505 resume_event: dict[str, Any] = {
506 "event": "loop_resume",
507 "ts": _iso_now(),
508 "loop": self.fsm.name,
509 "from_state": state.current_state,
510 "iteration": state.iteration,
511 }
512 if state.status == "awaiting_continuation" and state.continuation_prompt:
513 resume_event["from_handoff"] = True
514 resume_event["continuation_prompt"] = state.continuation_prompt
515 self.persistence.append_event(resume_event)
517 # Continue execution (don't clear previous events)
518 return self.run(clear_previous=False)
521def list_running_loops(loops_dir: Path | None = None) -> list[LoopState]:
522 """List all loops with saved state.
524 Args:
525 loops_dir: Base directory for loops (default: .loops)
527 Returns:
528 List of LoopState objects for all loops with state files
529 """
530 base_dir = loops_dir or Path(".loops")
531 running_dir = base_dir / RUNNING_DIR
533 if not running_dir.exists():
534 return []
536 states: list[LoopState] = []
537 for state_file in running_dir.glob("*.state.json"):
538 try:
539 data = json.loads(state_file.read_text())
540 states.append(LoopState.from_dict(data))
541 except (json.JSONDecodeError, KeyError):
542 continue # Skip malformed files
544 # Include loops that have a PID file but no state file yet (still starting up)
545 known_names = {s.loop_name for s in states}
546 for pid_file in running_dir.glob("*.pid"):
547 if pid_file.stem in known_names:
548 continue # state file already covers this loop
549 try:
550 pid = int(pid_file.read_text().strip())
551 except (ValueError, OSError):
552 continue
553 if _process_alive(pid):
554 states.append(
555 LoopState(
556 loop_name=pid_file.stem,
557 current_state="(initializing)",
558 iteration=0,
559 captured={},
560 prev_result=None,
561 last_result=None,
562 started_at="",
563 updated_at="",
564 status="starting",
565 )
566 )
568 return states
571def list_run_history(loop_name: str, loops_dir: Path | None = None) -> list[LoopState]:
572 """List archived runs for a loop, newest first.
574 Reads state files from .loops/.history/<run_id>-<loop_name>/state.json and
575 returns them sorted by started_at descending (most recent run first).
577 Also checks the legacy nested layout .loops/.history/<loop_name>/*/state.json
578 for backward compatibility with existing history folders.
580 Args:
581 loop_name: Name of the loop
582 loops_dir: Base directory for loops (default: .loops)
584 Returns:
585 List of LoopState objects for all archived runs, newest first.
586 Returns an empty list if no history exists.
587 """
588 base_dir = loops_dir or Path(".loops")
589 history_dir = base_dir / HISTORY_DIR
591 if not history_dir.exists():
592 return []
594 states: list[LoopState] = []
596 # Flat layout: <run_id>-<loop_name>/state.json
597 for state_file in history_dir.glob(f"*-{loop_name}/state.json"):
598 try:
599 data = json.loads(state_file.read_text())
600 states.append(LoopState.from_dict(data))
601 except (json.JSONDecodeError, KeyError):
602 continue
604 # Backward compat: legacy nested layout <loop_name>/<run_id>/state.json
605 old_loop_dir = history_dir / loop_name
606 if old_loop_dir.exists():
607 logger.warning(
608 "Found legacy nested history at %s; migrate to flat layout by moving "
609 "each run to .history/<run_id>-%s/",
610 old_loop_dir,
611 loop_name,
612 )
613 for state_file in old_loop_dir.glob("*/state.json"):
614 try:
615 data = json.loads(state_file.read_text())
616 states.append(LoopState.from_dict(data))
617 except (json.JSONDecodeError, KeyError):
618 continue
620 states.sort(key=lambda s: s.started_at, reverse=True)
621 return states
624def get_archived_events(
625 loop_name: str, run_id: str, loops_dir: Path | None = None
626) -> list[dict[str, Any]]:
627 """Read events for a specific archived run.
629 Args:
630 loop_name: Name of the loop
631 run_id: The run directory name (compact timestamp)
632 loops_dir: Base directory for loops (default: .loops)
634 Returns:
635 List of event dictionaries, empty if not found.
636 """
637 base_dir = loops_dir or Path(".loops")
638 run_folder = f"{run_id}-{loop_name}"
639 events_file = base_dir / HISTORY_DIR / run_folder / "events.jsonl"
641 if not events_file.exists():
642 return []
644 events: list[dict[str, Any]] = []
645 with open(events_file, encoding="utf-8") as f:
646 for line in f:
647 line = line.strip()
648 if line:
649 try:
650 events.append(json.loads(line))
651 except json.JSONDecodeError:
652 continue
653 return events
656def get_loop_history(loop_name: str, loops_dir: Path | None = None) -> list[dict[str, Any]]:
657 """Get event history for a loop.
659 Args:
660 loop_name: Name of the loop
661 loops_dir: Base directory for loops (default: .loops)
663 Returns:
664 List of event dictionaries
665 """
666 persistence = StatePersistence(loop_name, loops_dir)
667 return persistence.read_events()