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 collections import Counter
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
import echo  # noqa: E402
import echo_quality  # noqa: E402  — distinctive-token guards for fuzzy candidate matching

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")

# kind -> default `status:` stamped by capture when none is given. Every entity note is
# born with a status — a missing key is the #1 source of frontmatter drift (18/71 notes
# in the field audit). Records of past events default to done; living notes to active.
KIND_STATUS = {
    "person": "active", "company": "active", "concept": "active",
    "reference": "active", "meeting": "done", "project": "active",
    "area": "active", "semantic": "active", "episodic": "done",
    "working": "active", "skill": "active", "decision": "done",
}

# kind -> frontmatter fields that must be present AND non-empty on that kind's notes.
# Data-driven so vault_lint's incomplete-frontmatter check and sweep's backfill stay in
# lockstep with what capture stamps; add a (kind, field) pair here and all three follow.
KIND_REQUIRED_FM = {
    "person":    ("status", "tags"),
    "company":   ("status", "tags"),
    "concept":   ("status", "tags"),
    "reference": ("status", "tags"),
    "project":   ("status", "tags"),
    "area":      ("status", "tags"),
    "skill":     ("status", "tags"),
    "meeting":   ("tags",),
    "decision":  ("tags",),
}


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_aliases(title: str) -> list[str]:
    """Safe, lossless name variants of a title — case/punctuation normalizations that are
    unambiguously the SAME name (hyphen<->space, lowercased). Deliberately NOT acronyms or
    token subsets: those collide across entities and would make resolve() match too much.
    Lets `echo-memory` and `echo memory` both resolve to a title written either way."""
    t = str(title or "").strip()
    if not t:
        return []
    low = t.lower()
    variants = {low, slugify(t), slugify(t).replace("-", " "),
                low.replace("-", " "), low.replace(" ", "-")}
    return sorted(v for v in variants if v)


_WORD = re.compile(r"[a-z0-9]+")


def _sig_tokens(s) -> set[str]:
    """Distinctive tokens of a string: long enough and not a common word (reuses the
    auto-link guards), so 'data'/'api'/'the' can't drive a fuzzy match on their own."""
    return {t for t in _WORD.findall(str(s or "").lower())
            if len(t) >= echo_quality.MIN_NAME_LEN and t not in echo_quality._COMMON}


def aliases_in_frontmatter(text: str) -> list[str]:
    """Read an `aliases:` list (flow `[a, b]` or block `- a`) from a note's YAML frontmatter.
    Frontmatter is the durable, Obsidian-native home for aliases; sweep folds these into the
    index so they survive an index rebuild."""
    if not str(text).startswith("---"):
        return []
    end = text.find("\n---", 3)
    fm = text[:end] if end != -1 else text
    flow = re.search(r"(?m)^aliases:\s*\[(.*?)\]", fm)
    if flow:
        return [a.strip().strip('"').strip("'") for a in flow.group(1).split(",") if a.strip()]
    block = re.search(r"(?m)^aliases:\s*\n((?:[ \t]*-[ \t]*.+\n?)+)", fm)
    if block:
        return [ln.strip().lstrip("-").strip().strip('"').strip("'")
                for ln in block.group(1).splitlines() if ln.strip()]
    return []


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 EXACTLY matched by slug, title, or alias — else
    (None, None). Exact-only by design: this drives capture's create-vs-update decision, so a
    loose match here would silently merge distinct entities. Fuzzy lookup is fuzzy_candidates()."""
    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 fuzzy_candidates(index: dict, mention: str, limit: int = 5):
    """Advisory near-matches for when resolve() finds no exact hit. Ranks entities sharing
    >=1 *distinctive* token (>= MIN_NAME_LEN, not a common word) with the mention, so a
    shortened or expanded name — 'echo memory' for the project 'echo' — surfaces the existing
    note instead of vanishing. Returns [(slug, entry, score)] sorted best-first. NEVER used to
    auto-merge (that's resolve's job); only to SURFACE 'did you mean' candidates."""
    mtoks = _sig_tokens(mention)
    if not mtoks:
        return []
    scored = []
    for slug, e in index.get("entities", {}).items():
        etoks = set()
        for name in (slug, e.get("title", ""), *e.get("aliases", [])):
            etoks |= _sig_tokens(name)
        overlap = mtoks & etoks
        if not overlap or not etoks:
            continue
        score = round(len(overlap) / min(len(mtoks), len(etoks)), 3)
        scored.append((score, len(overlap), slug, e))
    scored.sort(key=lambda x: (-x[0], -x[1], x[2]))
    return [(slug, e, score) for score, _, slug, e in scored[:limit]]


def _entity_tokens(slug: str, e: dict) -> set[str]:
    toks: set[str] = set()
    for name in (slug, e.get("title", ""), *e.get("aliases", [])):
        toks |= _sig_tokens(name)
    return toks


def token_df(index: dict) -> Counter:
    """Vault-wide token document frequency: token -> how many entities carry it in their
    slug/title/aliases. A token shared by several entities ("echo" in an echo-centric
    vault) is a weak duplicate signal on its own; df lets the gate discount it."""
    df: Counter = Counter()
    for slug, e in index.get("entities", {}).items():
        for t in _entity_tokens(slug, e):
            df[t] += 1
    return df


def gate_candidates(index: dict, mention: str, kind: str | None = None,
                    threshold: float = 0.6):
    """The subset of fuzzy_candidates strong enough to BLOCK a create (capture's
    duplicate gate). Stricter than the advisory warning list, by two rules learned
    from live use (1.5.1):

      * same-kind only — a cross-kind name collision (a decision titled after its
        project, a meeting named for a company) is usually intentional naming, not a
        duplicate. Cross-kind lookalikes still surface as warnings, never as a block.
      * no single-common-token blocks — score = overlap/min(|tokens|) means an entity
        whose ONE distinctive token appears in the mention scores 1.0, so any title
        containing a vault-common word ("echo") would be blocked by every such entity.
        A lone shared token only blocks when it is unique to that entity (df == 1).

    Returns [(slug, entry, score)] like fuzzy_candidates."""
    mtoks = _sig_tokens(mention)
    if not mtoks:
        return []
    df = token_df(index)
    out = []
    for slug, e, score in fuzzy_candidates(index, mention):
        if score < threshold:
            continue
        if kind and e.get("kind") and e["kind"] != kind:
            continue
        overlap = mtoks & _entity_tokens(slug, e)
        if len(overlap) == 1 and df[next(iter(overlap))] > 1:
            continue
        out.append((slug, e, score))
    return out


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
    # Keep any prior + explicit aliases, and ALWAYS fold in the safe title variants so a
    # hyphen/space/case form of the name resolves even if no one passed --aliases. Drop any
    # alias that just equals the slug (resolve already checks the slug).
    merged = set(e.get("aliases", [])) | set(aliases or []) | set(derive_aliases(e["title"]))
    merged.discard(slug)
    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