| """ | |
| runtime_store.py — LibBee v3.1 | |
| Fixes applied: | |
| 1. save() now uses atomic write-then-rename (os.replace) to prevent file | |
| corruption on crash or mid-write failure. | |
| os.replace is atomic on POSIX (Linux/macOS) and effectively atomic on | |
| Windows (same filesystem). This protects feedback.jsonl, metrics.json, | |
| and runtime_config.json from partial writes. | |
| """ | |
| import json | |
| import os | |
| import time | |
| from pathlib import Path | |
| from typing import Any, Dict, Optional | |
| class JsonRuntimeStore: | |
| def __init__(self, path: Path, default: Optional[Dict[str, Any]] = None): | |
| self.path = Path(path) | |
| self.path.parent.mkdir(parents=True, exist_ok=True) | |
| self.default = default or {} | |
| def load(self) -> Dict[str, Any]: | |
| if not self.path.exists(): | |
| return dict(self.default) | |
| try: | |
| return json.loads(self.path.read_text(encoding="utf-8")) | |
| except Exception: | |
| return dict(self.default) | |
| def save(self, data: Dict[str, Any]) -> None: | |
| """Atomically write data to disk using a temp file + os.replace.""" | |
| tmp = self.path.with_suffix(".tmp") | |
| try: | |
| tmp.write_text( | |
| json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8" | |
| ) | |
| os.replace(tmp, self.path) | |
| except Exception: | |
| try: | |
| tmp.unlink(missing_ok=True) | |
| except Exception: | |
| pass | |
| raise | |
| def update(self, **kwargs: Any) -> Dict[str, Any]: | |
| data = self.load() | |
| data.update(kwargs) | |
| data["updated_at"] = time.time() | |
| self.save(data) | |
| return data | |