MP-1 (#368): Port PalaceRoom + Mempalace classes with 22 unit tests MP-2 (#369): L0-L5 retrieval order enforcer with recall-query detection MP-5 (#372): Wake-up protocol (300-900 token context), session scratchpad Modules: - mempalace.py: PalaceRoom + Mempalace dataclasses, factory constructors - retrieval_enforcer.py: Layered memory retrieval (identity → palace → scratch → gitea → skills) - wakeup.py: Session wake-up with caching (5min TTL) - scratchpad.py: JSON-based session notes with palace promotion All 65 tests pass. Pure stdlib + graceful degradation for ONNX issues (#373).
185 lines
5.3 KiB
Python
185 lines
5.3 KiB
Python
"""Session Scratchpad — ephemeral key-value notes per session.
|
|
|
|
Provides fast, JSON-backed scratch storage that lives for a session
|
|
and can be promoted to durable palace memory.
|
|
|
|
Storage: ~/.hermes/scratchpad/{session_id}.json
|
|
|
|
Refs: Epic #367, Sub-issue #372
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Constants
|
|
# ---------------------------------------------------------------------------
|
|
|
|
SCRATCHPAD_DIR = Path.home() / ".hermes" / "scratchpad"
|
|
MEMPALACE_BIN = "/Library/Frameworks/Python.framework/Versions/3.12/bin/mempalace"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Internal helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _scratch_path(session_id: str) -> Path:
|
|
"""Return the JSON file path for a given session."""
|
|
# Sanitize session_id to prevent path traversal
|
|
safe_id = "".join(c for c in session_id if c.isalnum() or c in "-_")
|
|
if not safe_id:
|
|
safe_id = "unnamed"
|
|
return SCRATCHPAD_DIR / f"{safe_id}.json"
|
|
|
|
|
|
def _load(session_id: str) -> dict:
|
|
"""Load scratchpad data, returning empty dict on failure."""
|
|
path = _scratch_path(session_id)
|
|
try:
|
|
if path.exists():
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
pass
|
|
return {}
|
|
|
|
|
|
def _save(session_id: str, data: dict) -> None:
|
|
"""Persist scratchpad data to disk."""
|
|
SCRATCHPAD_DIR.mkdir(parents=True, exist_ok=True)
|
|
path = _scratch_path(session_id)
|
|
path.write_text(json.dumps(data, indent=2, default=str), encoding="utf-8")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Public API
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def write_scratch(session_id: str, key: str, value: Any) -> None:
|
|
"""Write a note to the session scratchpad.
|
|
|
|
Args:
|
|
session_id: Current session identifier.
|
|
key: Note key (string).
|
|
value: Note value (any JSON-serializable type).
|
|
"""
|
|
data = _load(session_id)
|
|
data[key] = {
|
|
"value": value,
|
|
"written_at": time.strftime("%Y-%m-%d %H:%M:%S"),
|
|
}
|
|
_save(session_id, data)
|
|
|
|
|
|
def read_scratch(session_id: str, key: Optional[str] = None) -> dict:
|
|
"""Read session scratchpad (all keys or one).
|
|
|
|
Args:
|
|
session_id: Current session identifier.
|
|
key: Optional specific key. If None, returns all entries.
|
|
|
|
Returns:
|
|
dict — either {key: {value, written_at}} or the full scratchpad.
|
|
"""
|
|
data = _load(session_id)
|
|
if key is not None:
|
|
entry = data.get(key)
|
|
return {key: entry} if entry else {}
|
|
return data
|
|
|
|
|
|
def delete_scratch(session_id: str, key: str) -> bool:
|
|
"""Remove a single key from the scratchpad.
|
|
|
|
Returns True if the key existed and was removed.
|
|
"""
|
|
data = _load(session_id)
|
|
if key in data:
|
|
del data[key]
|
|
_save(session_id, data)
|
|
return True
|
|
return False
|
|
|
|
|
|
def list_sessions() -> list[str]:
|
|
"""List all session IDs that have scratchpad files."""
|
|
try:
|
|
if SCRATCHPAD_DIR.exists():
|
|
return [
|
|
f.stem
|
|
for f in SCRATCHPAD_DIR.iterdir()
|
|
if f.suffix == ".json" and f.is_file()
|
|
]
|
|
except OSError:
|
|
pass
|
|
return []
|
|
|
|
|
|
def promote_to_palace(
|
|
session_id: str,
|
|
key: str,
|
|
room: str = "general",
|
|
drawer: Optional[str] = None,
|
|
) -> bool:
|
|
"""Move a scratchpad note to durable palace memory.
|
|
|
|
Uses the mempalace CLI to store the note in the specified room.
|
|
Removes the note from the scratchpad after successful promotion.
|
|
|
|
Args:
|
|
session_id: Session containing the note.
|
|
key: Scratchpad key to promote.
|
|
room: Palace room name (default: 'general').
|
|
drawer: Optional drawer name within the room. Defaults to key.
|
|
|
|
Returns:
|
|
True if promotion succeeded, False otherwise.
|
|
"""
|
|
data = _load(session_id)
|
|
entry = data.get(key)
|
|
if not entry:
|
|
return False
|
|
|
|
value = entry.get("value", entry) if isinstance(entry, dict) else entry
|
|
content = json.dumps(value, default=str) if not isinstance(value, str) else value
|
|
|
|
try:
|
|
bin_path = MEMPALACE_BIN if os.path.exists(MEMPALACE_BIN) else "mempalace"
|
|
target_drawer = drawer or key
|
|
result = subprocess.run(
|
|
[bin_path, "store", room, target_drawer, content],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
if result.returncode == 0:
|
|
# Remove from scratchpad after successful promotion
|
|
del data[key]
|
|
_save(session_id, data)
|
|
return True
|
|
except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
|
|
# mempalace CLI not available — degrade gracefully
|
|
pass
|
|
|
|
return False
|
|
|
|
|
|
def clear_session(session_id: str) -> bool:
|
|
"""Delete the entire scratchpad for a session.
|
|
|
|
Returns True if the file existed and was removed.
|
|
"""
|
|
path = _scratch_path(session_id)
|
|
try:
|
|
if path.exists():
|
|
path.unlink()
|
|
return True
|
|
except OSError:
|
|
pass
|
|
return False
|