feat(memory): add JSON index and runtime memory_lookup/read/write tools

Gives Pyra an active memory brain: memory_index.json tracks summary +
keywords per file (like an inode table), and three built-in tools let
the AI look up, read, and overwrite memory mid-session. write_memory
accepts summary/keywords; update_index() merges the JSON index without
losing existing metadata.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
curo1305
2026-05-17 23:01:54 +02:00
parent ad024807bc
commit e56e9779ec
4 changed files with 163 additions and 7 deletions
+44 -4
View File
@@ -1,22 +1,51 @@
import datetime
import json
from pathlib import Path
from pyra.memory import _MEMORY_ROOT
from pyra.utils.paths import safe_chmod
_INDEX_FILE = _MEMORY_ROOT / "MEMORY_INDEX.md"
_JSON_INDEX_FILE = _MEMORY_ROOT / "memory_index.json"
_EXCLUDED = {"MEMORY_INDEX.md", "memory_index.json"}
def _load_json_index() -> dict:
if not _JSON_INDEX_FILE.exists():
return {}
try:
return json.loads(_JSON_INDEX_FILE.read_text())
except (json.JSONDecodeError, OSError):
return {}
def update_index() -> None:
files = sorted(_MEMORY_ROOT.rglob("*.md"))
files = [f for f in files if f.name != "MEMORY_INDEX.md"]
existing = _load_json_index()
files = sorted(_MEMORY_ROOT.rglob("*.md"))
files = [f for f in files if f.name not in _EXCLUDED]
new_json: dict = {}
rows: list[str] = []
for f in files:
rel = f.relative_to(_MEMORY_ROOT)
rel_key = rel.as_posix()
category = rel.parts[0] if len(rel.parts) > 1 else "root"
mtime = datetime.datetime.fromtimestamp(f.stat().st_mtime).strftime("%Y-%m-%d %H:%M")
rows.append(f"| {rel} | {category} | {mtime} |")
mtime = datetime.datetime.fromtimestamp(f.stat().st_mtime)
mtime_str = mtime.strftime("%Y-%m-%d %H:%M")
prev = existing.get(rel_key, {})
new_json[rel_key] = {
"summary": prev.get("summary", ""),
"keywords": prev.get("keywords", []),
"modified": mtime.isoformat(timespec="seconds"),
}
rows.append(f"| {rel} | {category} | {mtime_str} |")
_JSON_INDEX_FILE.write_text(json.dumps(new_json, indent=2))
safe_chmod(_JSON_INDEX_FILE, 0o600)
table = "\n".join(rows) if rows else "| _(no memory files)_ | — | — |"
content = (
@@ -28,3 +57,14 @@ def update_index() -> None:
)
_INDEX_FILE.write_text(content)
safe_chmod(_INDEX_FILE, 0o600)
def update_json_entry(rel_path: str, summary: str, keywords: list[str]) -> None:
"""Update the summary and keywords for one entry in the JSON index."""
index = _load_json_index()
entry = index.get(rel_path, {})
entry["summary"] = summary
entry["keywords"] = keywords
index[rel_path] = entry
_JSON_INDEX_FILE.write_text(json.dumps(index, indent=2))
safe_chmod(_JSON_INDEX_FILE, 0o600)
+29
View File
@@ -1,10 +1,13 @@
import datetime
import json
from dataclasses import dataclass
from pathlib import Path
from pyra.memory import _MEMORY_ROOT
from pyra.security.boundaries import assert_safe_path
_JSON_INDEX_FILE = _MEMORY_ROOT / "memory_index.json"
@dataclass
class MemoryFile:
@@ -49,6 +52,32 @@ def read_memory(name: str) -> str:
return path.read_text()
def read_index() -> dict:
"""Return memory_index.json contents, or {} if missing or corrupt."""
if not _JSON_INDEX_FILE.exists():
return {}
try:
return json.loads(_JSON_INDEX_FILE.read_text())
except (json.JSONDecodeError, OSError):
return {}
def lookup_memories(query: str) -> list[dict]:
"""Case-insensitive substring search over summary text and keywords."""
q = query.lower()
results: list[dict] = []
for rel_path, entry in read_index().items():
summary = entry.get("summary", "").lower()
keywords = [k.lower() for k in entry.get("keywords", [])]
if q in summary or any(q in k or k in q for k in keywords):
results.append({
"file": rel_path,
"summary": entry.get("summary", ""),
"keywords": entry.get("keywords", []),
})
return results
def load_context_for_session() -> str:
memories = list_memories()
if not memories:
+10 -2
View File
@@ -1,7 +1,7 @@
from pathlib import Path
from pyra.memory import _MEMORY_ROOT
from pyra.memory.index import update_index
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
@@ -20,12 +20,20 @@ def _resolve_and_validate(name: str) -> Path:
return path
def write_memory(name: str, content: str) -> Path:
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 [])
return path