chorus/echo-memory.plugin.src/skills/echo-memory/scripts/echo_index.py

#!/usr/bin/env python3
"""echo_index.py — the ECHO entity index.

A small machine-readable registry at `_agent/index/entities.json` that turns
routing/resolve into an O(1) lookup (no fuzzy search per write) and supplies the
name->path map used for cross-linking and recall.

  {
    "schema": 1,
    "updated": "YYYY-MM-DD",
    "entities": {
      "<slug>": {"path": "...", "kind": "...", "title": "...",
                 "aliases": [...], "last_seen": "YYYY-MM-DD"}
    }
  }

Imported by echo_ops.py, sweep.py, and vault_lint.py. Network goes through echo.py.
"""

from __future__ import annotations

import json
import re
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
import echo  # noqa: E402

INDEX_PATH = "_agent/index/entities.json"

# kind -> destination folder (the canonical home for that entity kind)
KIND_FOLDER = {
    "person":    "resources/people",
    "company":   "resources/companies",
    "concept":   "resources/concepts",
    "reference": "resources/references",
    "meeting":   "resources/meetings",
    "project":   "projects/active",            # new projects default to active
    "area":      "areas",                      # needs a domain segment
    "semantic":  "_agent/memory/semantic",
    "episodic":  "_agent/memory/episodic",
    "working":   "_agent/memory/working",
    "skill":     "_agent/skills/active",
    "decision":  "decisions/by-date",
}
# kind -> frontmatter `type:` value
KIND_TYPE = {
    "person": "person", "company": "company", "concept": "concept",
    "reference": "reference", "meeting": "meeting", "project": "project",
    "area": "area", "semantic": "semantic-memory", "episodic": "episodic-memory",
    "working": "working-memory", "skill": "skill", "decision": "decision",
}
DATE_KINDS = {"meeting", "decision"}          # filename carries a YYYY-MM-DD prefix
AREA_DOMAINS = ("business", "personal", "learning", "systems")


def slugify(text: str) -> str:
    text = str(text).strip().lower()
    text = re.sub(r"[^a-z0-9]+", "-", text).strip("-")
    return text[:40] or "untitled"


def derive_path(kind: str, slug: str, date: str | None = None, domain: str = "business") -> str:
    folder = KIND_FOLDER.get(kind)
    if folder is None:
        raise echo.EchoError(f"unknown kind '{kind}' (one of: {', '.join(sorted(KIND_FOLDER))})", 2)
    if kind == "area":
        if domain not in AREA_DOMAINS:
            raise echo.EchoError(f"area domain must be one of {AREA_DOMAINS}", 2)
        return f"areas/{domain}/{slug}.md"
    if kind in DATE_KINDS:
        return f"{folder}/{date or echo.today()}-{slug}.md"
    return f"{folder}/{slug}.md"


_KIND_RULES = [
    (r"^resources/people/", "person"),
    (r"^resources/companies/", "company"),
    (r"^resources/concepts/", "concept"),
    (r"^resources/references/", "reference"),
    (r"^resources/meetings/", "meeting"),
    (r"^projects/(active|incubating|on-hold|archived)/", "project"),
    (r"^areas/[^/]+/", "area"),
    (r"^decisions/by-date/", "decision"),
    (r"^_agent/memory/semantic/", "semantic"),
    (r"^_agent/memory/episodic/", "episodic"),
    (r"^_agent/skills/(active|archived)/", "skill"),
]


def kind_for_path(path: str) -> str | None:
    """The entity kind a vault path belongs to (the inverse of derive_path), or None
    if the path is not an indexable entity note (journal, sessions, inbox, ...)."""
    for rx, kind in _KIND_RULES:
        if re.match(rx, path):
            return kind
    return None


def empty_index() -> dict:
    return {"schema": 1, "updated": echo.today(), "entities": {}}


def load() -> dict:
    status, body = echo.request("GET", echo.vault_url(INDEX_PATH))
    if status == 404:
        return empty_index()
    echo.check(status, body, "index load")
    try:
        d = json.loads(body)
    except json.JSONDecodeError:
        return empty_index()
    d.setdefault("entities", {})
    return d


def save(index: dict) -> None:
    index["updated"] = echo.today()
    body = json.dumps(index, indent=2, ensure_ascii=False).encode("utf-8")
    status, b = echo.request("PUT", echo.vault_url(INDEX_PATH), data=body,
                             headers={"Content-Type": "application/json"})
    echo.check(status, b, "index save")


def _norm(s) -> str:
    return slugify(s)


def resolve(index: dict, mention: str):
    """Return (slug, entry) for a mention matched by slug, title, or alias — else (None, None)."""
    key = _norm(mention)
    ents = index.get("entities", {})
    if key in ents:
        return key, ents[key]
    plain = str(mention).strip().lower()
    for slug, e in ents.items():
        cands = {slug, _norm(e.get("title", "")), str(e.get("title", "")).strip().lower()}
        cands |= {_norm(a) for a in e.get("aliases", [])}
        cands |= {str(a).strip().lower() for a in e.get("aliases", [])}
        if key in cands or plain in cands:
            return slug, e
    return None, None


def upsert(index: dict, slug: str, path: str, kind: str, title: str | None = None,
           aliases=None) -> dict:
    ents = index.setdefault("entities", {})
    e = ents.get(slug, {})
    e["path"] = path
    e["kind"] = kind
    e["title"] = title or e.get("title") or slug
    merged = set(e.get("aliases", [])) | set(aliases or [])
    e["aliases"] = sorted(a for a in merged if a)
    e["last_seen"] = echo.today()
    ents[slug] = e
    return e


def name_map(index: dict) -> dict:
    """Map every resolvable name -> path: slug, normalized title, basename, aliases.
    Used to resolve `[[wikilinks]]` and entity mentions back to vault paths."""
    m: dict[str, str] = {}
    for slug, e in index.get("entities", {}).items():
        path = e.get("path")
        if not path:
            continue
        base = path.rsplit("/", 1)[-1]
        base = base[:-3] if base.endswith(".md") else base
        keys = {slug, _norm(e.get("title", "")), _norm(base)}
        keys |= {_norm(a) for a in e.get("aliases", [])}
        for k in keys:
            if k:
                m.setdefault(k, path)
    return m