Save memory to file atomically
(memory_data: Dict[str, Any], memory_file: str)
| 31 | |
| 32 | |
| 33 | async def _save_memory(memory_data: Dict[str, Any], memory_file: str) -> bool: |
| 34 | """Save memory to file atomically""" |
| 35 | temp_file = f"{memory_file}.tmp" |
| 36 | |
| 37 | try: |
| 38 | # Ensure directory exists |
| 39 | Path(memory_file).parent.mkdir(parents=True, exist_ok=True) |
| 40 | |
| 41 | # Write to temporary file first (atomic operation) |
| 42 | with open(temp_file, "w", encoding="utf-8") as f: |
| 43 | json.dump(memory_data, f, indent=2, ensure_ascii=False) |
| 44 | |
| 45 | # Atomic move |
| 46 | os.rename(temp_file, memory_file) |
| 47 | return True |
| 48 | |
| 49 | except Exception as e: |
| 50 | print(f"Error: Could not save memory file {memory_file}: {e}") |
| 51 | # Clean up temp file if it exists |
| 52 | if os.path.exists(temp_file): |
| 53 | try: |
| 54 | os.remove(temp_file) |
| 55 | except Exception: |
| 56 | pass |
| 57 | return False |
| 58 | |
| 59 | |
| 60 | async def memory( |
no outgoing calls