import datetime from pathlib import Path from pyra.memory import _MEMORY_ROOT from pyra.memory.index import update_index, update_json_entry from pyra.security.boundaries import assert_safe_path from pyra.utils.paths import safe_chmod def _resolve_and_validate(name: str) -> Path: if name.startswith("/") or name.startswith("~"): raise ValueError(f"Memory name must be relative, got: {name!r}") path = (_MEMORY_ROOT / name).resolve() assert_safe_path(path) try: path.relative_to(_MEMORY_ROOT.resolve()) except ValueError: raise PermissionError(f"Path escapes memory directory: {name!r}") if path.suffix.lower() != ".md": path = path.with_suffix(".md") return path def _upsert_to_db(path: Path, content: str, summary: str = "", keywords: list[str] | None = None) -> None: from pyra.memory import database if not database._DB_PATH.exists(): return rel = path.relative_to(_MEMORY_ROOT).as_posix() category = rel.split("/")[0] if "/" in rel else "root" stat = path.stat() mtime = datetime.datetime.fromtimestamp(stat.st_mtime).isoformat(timespec="seconds") database.upsert( rel, content=content, category=category, size_bytes=stat.st_size, modified=mtime, summary=summary, keywords=keywords, ) def write_memory( name: str, content: str, summary: str = "", keywords: list[str] | None = None, ) -> Path: path = _resolve_and_validate(name) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content) safe_chmod(path, 0o600) update_index() if summary or keywords: rel_key = path.relative_to(_MEMORY_ROOT).as_posix() update_json_entry(rel_key, summary, keywords or []) _upsert_to_db(path, content, summary, keywords) return path def append_memory(name: str, content: str) -> Path: path = _resolve_and_validate(name) path.parent.mkdir(parents=True, exist_ok=True) if path.exists(): existing = path.read_text() new_content = existing.rstrip() + "\n\n" + content path.write_text(new_content) else: new_content = content path.write_text(new_content) safe_chmod(path, 0o600) update_index() _upsert_to_db(path, new_content) return path