Files
Cortex-Inara/cortex/session_store.py
Scott Idem 8add4ffd02 Add edit/delete history, named sessions, scroll fix, systemd service
- Edit/delete individual messages from session context with inline editing
  (Ctrl+Enter saves, Escape cancels); changes sync to backend via PUT /history
- PUT /history/{session_id} endpoint to replace full message list
- Named sessions: readable slugs (e.g. quiet-spring) instead of UUID fragments
- Scroll no longer snaps to bottom when user has scrolled up to read history
- cortex.service: systemd unit for auto-start and restart-on-failure

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-10 23:38:39 -04:00

88 lines
2.9 KiB
Python

import json
import random
from pathlib import Path
from datetime import datetime
from config import settings
_ADJECTIVES = [
"amber", "azure", "bold", "bright", "calm", "cedar", "cobalt", "coral",
"crisp", "dusk", "ember", "fern", "frost", "gold", "hazy", "indigo",
"jade", "keen", "lark", "lunar", "maple", "misty", "noble", "north",
"oak", "onyx", "opal", "pine", "quiet", "raven", "ridge", "river",
"sage", "silver", "slate", "solar", "steel", "stone", "swift", "teal",
"timber", "vale", "velvet", "violet", "warm", "wild", "winter", "wren",
]
_NOUNS = [
"bay", "bloom", "brook", "canyon", "cave", "cliff", "cloud", "coast",
"creek", "dale", "dawn", "delta", "dune", "echo", "falls", "field",
"fjord", "flare", "glade", "glen", "grove", "harbor", "haven", "hill",
"hollow", "isle", "lake", "ledge", "marsh", "meadow", "mist", "moon",
"moor", "peak", "plain", "pond", "reef", "ridge", "rise", "rock",
"shore", "sky", "slope", "spring", "star", "storm", "trail", "vale",
"wave", "wood",
]
def generate_session_id() -> str:
"""Return a readable slug like 'amber-lake' or 'amber-lake-03'."""
existing = {s["session_id"] for s in list_all()}
for _ in range(30):
base = f"{random.choice(_ADJECTIVES)}-{random.choice(_NOUNS)}"
if base not in existing:
return base
# Try with a numeric suffix
candidate = f"{base}-{random.randint(2, 99):02d}"
if candidate not in existing:
return candidate
# Statistically unreachable for a personal tool
import uuid
return str(uuid.uuid4())[:8]
def _path(session_id: str) -> Path:
d = settings.sessions_path()
d.mkdir(parents=True, exist_ok=True)
return d / f"{session_id}.json"
def load(session_id: str) -> list[dict]:
path = _path(session_id)
if not path.exists():
return []
return json.loads(path.read_text()).get("messages", [])
def save(session_id: str, messages: list[dict]) -> None:
path = _path(session_id)
existing = json.loads(path.read_text()) if path.exists() else {}
# Enforce rolling window
windowed = messages[-settings.max_history_messages:]
path.write_text(json.dumps({
"session_id": session_id,
"created": existing.get("created", datetime.now().isoformat()),
"updated": datetime.now().isoformat(),
"messages": windowed,
}, indent=2))
def list_all() -> list[dict]:
d = settings.sessions_path()
if not d.exists():
return []
results = []
for f in sorted(d.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True):
try:
data = json.loads(f.read_text())
results.append({
"session_id": data["session_id"],
"updated": data.get("updated"),
"message_count": len(data.get("messages", [])),
})
except Exception:
pass
return results