54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
"""Atomic loop checkpoint management.
|
|
|
|
Issue #5 Phase 3.
|
|
Saves and restores loop progress (.mam/loop-checkpoint.json) across
|
|
signal-triggered quiesce or resumes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
|
|
def save_checkpoint(checkpoint_path: str | Path, data: Dict[str, Any]) -> None:
|
|
"""Atomically write loop state to a checkpoint file."""
|
|
path = Path(checkpoint_path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp_path = path.with_suffix(f".tmp.{os.getpid()}")
|
|
try:
|
|
with open(tmp_path, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=2)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
tmp_path.replace(path)
|
|
finally:
|
|
if tmp_path.exists():
|
|
try:
|
|
tmp_path.unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def load_checkpoint(checkpoint_path: str | Path) -> Optional[Dict[str, Any]]:
|
|
"""Load loop state from a checkpoint file if present, else None."""
|
|
path = Path(checkpoint_path)
|
|
if not path.is_file():
|
|
return None
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except (OSError, json.JSONDecodeError):
|
|
return None
|
|
|
|
|
|
def remove_checkpoint(checkpoint_path: str | Path) -> None:
|
|
"""Safely remove a checkpoint file."""
|
|
path = Path(checkpoint_path)
|
|
try:
|
|
path.unlink(missing_ok=True)
|
|
except OSError:
|
|
pass
|