neuron/tools/retrieval-eval/run_eval.py

#!/usr/bin/env python3
"""
run_eval.py — measure one soul build's retrieval against the gold set.

WHAT IT MEASURES, AND WHY IT BOOTS A REAL SOUL
    The point is Will's designed retrieval — spreading activation over the
    weighted directed graph with four-factor multiplicative scoring — not a
    Python re-implementation of it. A re-implementation would measure my
    reading of the design; booting the compiled binary measures the design. So
    this harness compiles the actual `soul.el` amalgam (build-soul.sh) and asks
    it over HTTP, exactly as the MCP wrapper and the app do.

SAFETY — read this before changing anything here
    * Boots on a THROWAWAY port with a THROWAWAY $HOME and a THROWAWAY COPY of
      the corpus. Refuses to use 7770 / 8742 / 7779 / 17779 / 7771.
    * ENGRAM_URL / SOUL_ENGRAM_URL are UNSET and SOUL_ISE_URL is pinned to a
      dead port. This is not belt-and-braces: the periodic engram sync resolves
      its source as env(SOUL_ISE_URL) -> state -> DEFAULT http://localhost:8742,
      so leaving it unset makes an "isolated" run silently pull the operator's
      LIVE brain. (Learned the hard way on 2026-08-03; see the same note in
      scripts/verify-soul-contract.sh.)
    * Every process this file starts is tracked and killed in a finally block,
      then CONFIRMED dead by pid probe, and the confirmation is written into the
      results file. A run that cannot confirm its child is dead exits non-zero.
    * Activation is a STATEFUL read by design (patent claim 29: traversal
      updates last-activation and increments activation counts). The corpus copy
      is therefore per-run and disposable, and every run starts from a byte-
      identical copy so two configurations see the same starting graph.

usage:
  python3 run_eval.py --soul <binary> --corpus <snapshot.json> --label main \
      [--port 7893] [--gold gold_set.json] [--limit 10] [--out results-main.json]
"""
import argparse
import json
import os
import shutil
import signal
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request

HERE = os.path.dirname(os.path.abspath(__file__))
FORBIDDEN_PORTS = {7770, 8742, 7779, 17779, 7771, 8080}


# ─────────────────────────────────────────────────────────────────────────────
# metrics
# ─────────────────────────────────────────────────────────────────────────────
def recall_at_k(returned, relevant, k):
    if not relevant:
        return None
    return len(set(returned[:k]) & set(relevant)) / len(relevant)


def hit_at_k(returned, relevant, k):
    if not relevant:
        return None
    return 1.0 if set(returned[:k]) & set(relevant) else 0.0


def precision_at_k(returned, relevant, k):
    """Fixed denominator k, as in docs/research/graphrag_eval/score.py.

    Fixed denominator penalises an empty result and a page of junk equally,
    which is what we want: a retriever that returns nothing is not 'precise'.
    """
    if not relevant:
        return None
    return len(set(returned[:k]) & set(relevant)) / k


def mrr(returned, relevant, k):
    if not relevant:
        return None
    rel = set(relevant)
    for i, nid in enumerate(returned[:k], start=1):
        if nid in rel:
            return 1.0 / i
    return 0.0


def mean(vals):
    vals = [v for v in vals if v is not None]
    return sum(vals) / len(vals) if vals else 0.0


def pct(vals):
    return f"{100 * mean(vals):.1f}%"


# ─────────────────────────────────────────────────────────────────────────────
# soul lifecycle
# ─────────────────────────────────────────────────────────────────────────────
class Soul:
    def __init__(self, binary, corpus, port, verbose=True):
        if port in FORBIDDEN_PORTS:
            raise SystemExit(f"REFUSING: port {port} is a live service port.")
        self.binary = os.path.abspath(binary)
        self.corpus = os.path.abspath(corpus)
        self.port = port
        self.verbose = verbose
        self.home = None
        self.proc = None
        self.pid = None
        self.log = None
        self.confirmed_dead = None

    @property
    def base(self):
        return f"http://127.0.0.1:{self.port}"

    def start(self, boot_timeout=180):
        self.home = tempfile.mkdtemp(prefix="retrieval-eval-home.")
        snap = os.path.join(self.home, "corpus.json")
        t0 = time.time()
        shutil.copyfile(self.corpus, snap)   # per-run disposable copy, never the source
        self.log = os.path.join(self.home, "soul.log")

        env = {k: v for k, v in os.environ.items()
               if k not in ("ENGRAM_URL", "ENGRAM_API_KEY", "SOUL_ENGRAM_URL",
                            "ANTHROPIC_API_KEY", "NEURON_LLM_API_KEY", "SOUL_IDENTITY",
                            "SOUL_API_KEY")}
        env.update({
            "HOME": self.home,
            "NEURON_PORT": str(self.port),
            "SOUL_CGI_ID": f"ntn-retrieval-eval-{os.getpid()}",
            "SOUL_ENGRAM_PATH": snap,
            "NEURON_API_URL": "http://127.0.0.1:9",   # dead port
            "SOUL_ISE_URL": "http://127.0.0.1:9",     # dead port — see SAFETY above
            # Park the background loops for an hour so heartbeat/consolidation
            # cannot mutate the graph between queries and make runs unrepeatable.
            "SOUL_TICK_MS": "3600000",
            "SOUL_HEARTBEAT_MS": "3600000",
            "SOUL_REFRESH_MS": "3600000",
        })
        with open(self.log, "wb") as lf:
            self.proc = subprocess.Popen([self.binary], env=env, stdout=lf, stderr=lf,
                                         start_new_session=True)
        self.pid = self.proc.pid
        if self.verbose:
            print(f"  booted pid={self.pid} port={self.port} home={self.home}")

        deadline = time.time() + boot_timeout
        while time.time() < deadline:
            if self.proc.poll() is not None:
                raise RuntimeError(f"soul exited during boot: {self._log_tail()}")
            rss = self._rss_kb()
            if rss and rss > 6 * 1024 * 1024:
                self.stop()
                raise RuntimeError(f"soul RSS {rss}KB > 6GB — aborted")
            try:
                with urllib.request.urlopen(f"{self.base}/health", timeout=2) as r:
                    if r.status == 200:
                        if self.verbose:
                            print(f"  healthy in {time.time() - t0:.1f}s, RSS={self._rss_kb()}KB")
                        return
            except Exception:
                pass
            time.sleep(0.5)
        self.stop()
        raise RuntimeError(f"soul never healthy on {self.base}: {self._log_tail()}")

    def _rss_kb(self):
        try:
            out = subprocess.run(["ps", "-o", "rss=", "-p", str(self.pid)],
                                 capture_output=True, text=True, timeout=5).stdout.strip()
            return int(out) if out else None
        except Exception:
            return None

    def _log_tail(self, n=15):
        try:
            with open(self.log, encoding="utf-8", errors="replace") as fh:
                return "\n".join(fh.read().splitlines()[-n:])
        except Exception:
            return "(no log)"

    def recall(self, query, limit, timeout=60):
        url = f"{self.base}/api/neuron/recall?query={urllib.parse.quote(query)}&limit={limit}"
        t0 = time.perf_counter()
        try:
            with urllib.request.urlopen(url, timeout=timeout) as r:
                raw = r.read().decode("utf-8", "replace")
            ms = (time.perf_counter() - t0) * 1000
        except Exception as exc:
            return [], (time.perf_counter() - t0) * 1000, f"{type(exc).__name__}: {exc}"
        try:
            arr = json.loads(raw)
        except Exception:
            return [], ms, f"unparseable response ({len(raw)}B)"
        if not isinstance(arr, list):
            return [], ms, f"non-array response: {str(arr)[:120]}"
        ids = [x.get("id") for x in arr if isinstance(x, dict) and x.get("id")]
        return ids, ms, None

    def stop(self):
        """Kill and CONFIRM. A test process that outlives its test is a bug."""
        if self.pid is None:
            self.confirmed_dead = True
            return True
        for sig in (signal.SIGTERM, signal.SIGKILL):
            try:
                os.kill(self.pid, sig)
            except ProcessLookupError:
                break
            except Exception:
                pass
            for _ in range(20):
                try:
                    os.kill(self.pid, 0)
                except ProcessLookupError:
                    break
                time.sleep(0.1)
            else:
                continue
            break
        try:
            self.proc.wait(timeout=5)
        except Exception:
            pass
        try:
            os.kill(self.pid, 0)
            self.confirmed_dead = False
        except ProcessLookupError:
            self.confirmed_dead = True
        if self.verbose:
            print(f"  pid {self.pid}: {'CONFIRMED DEAD' if self.confirmed_dead else 'STILL ALIVE'}")
        if self.home and os.path.isdir(self.home):
            shutil.rmtree(self.home, ignore_errors=True)
        return self.confirmed_dead


# ─────────────────────────────────────────────────────────────────────────────
# eval
# ─────────────────────────────────────────────────────────────────────────────
def evaluate(soul, gold, limit):
    rows = []
    for q in gold["queries"]:
        ids, ms, err = soul.recall(q["query"], limit)
        rel = q.get("relevant") or []
        row = {
            "id": q["id"],
            "category": q["category"],
            "query": q["query"],
            "returned": ids,
            "n_returned": len(ids),
            "latency_ms": round(ms, 1),
            "error": err,
        }
        if q.get("expect_empty"):
            row["clean"] = (len(ids) == 0)
            row["false_positives"] = len(ids)
        else:
            row["hit@5"] = hit_at_k(ids, rel, 5)
            row["recall@5"] = recall_at_k(ids, rel, 5)
            row["recall@10"] = recall_at_k(ids, rel, 10)
            row["precision@5"] = precision_at_k(ids, rel, 5)
            row["mrr@10"] = mrr(ids, rel, 10)
        if q.get("must_outrank"):
            correct, stale = q["must_outrank"]
            ic = ids.index(correct) if correct in ids else None
            istale = ids.index(stale) if stale in ids else None
            # Correct must be present AND above the stale node. A run that
            # returns neither is NOT a pass: the corrected fact is what the
            # user needed.
            row["outranks"] = (ic is not None) and (istale is None or ic < istale)
            row["rank_correct"] = None if ic is None else ic + 1
            row["rank_stale"] = None if istale is None else istale + 1
        rows.append(row)
    return rows


def aggregate(rows):
    scored = [r for r in rows if "hit@5" in r]
    nonsense = [r for r in rows if "clean" in r]
    outrank = [r for r in rows if "outranks" in r]
    lat = sorted(r["latency_ms"] for r in rows)
    agg = {
        "n_queries": len(rows),
        "n_scored": len(scored),
        "hit@5": mean([r["hit@5"] for r in scored]),
        "recall@5": mean([r["recall@5"] for r in scored]),
        "recall@10": mean([r["recall@10"] for r in scored]),
        "precision@5": mean([r["precision@5"] for r in scored]),
        "mrr@10": mean([r["mrr@10"] for r in scored]),
        "nonsense_clean": f"{sum(1 for r in nonsense if r['clean'])}/{len(nonsense)}",
        "superseded_outranks": f"{sum(1 for r in outrank if r['outranks'])}/{len(outrank)}",
        "latency_ms_p50": lat[len(lat) // 2] if lat else 0,
        "latency_ms_p95": lat[max(0, int(len(lat) * 0.95) - 1)] if lat else 0,
        "latency_ms_max": lat[-1] if lat else 0,
        "errors": sum(1 for r in rows if r["error"]),
        "by_category": {},
    }
    cats = sorted({r["category"] for r in rows})
    for c in cats:
        cr = [r for r in rows if r["category"] == c]
        if c == "nonsense":
            agg["by_category"][c] = {
                "n": len(cr),
                "clean": sum(1 for r in cr if r["clean"]),
                "avg_false_positives": mean([float(r["false_positives"]) for r in cr]),
            }
        else:
            e = {
                "n": len(cr),
                "hit@5": mean([r.get("hit@5") for r in cr]),
                "recall@5": mean([r.get("recall@5") for r in cr]),
                "recall@10": mean([r.get("recall@10") for r in cr]),
                "mrr@10": mean([r.get("mrr@10") for r in cr]),
            }
            if c == "superseded":
                e["outranks"] = sum(1 for r in cr if r.get("outranks"))
            agg["by_category"][c] = e
    return agg


def print_table(label, agg):
    print(f"\n=== {label} ===")
    print(f"  queries {agg['n_queries']} ({agg['n_scored']} scored + "
          f"{agg['n_queries'] - agg['n_scored']} control) · errors {agg['errors']}")
    print(f"  {'hit@5':>12} {'recall@5':>10} {'recall@10':>10} {'prec@5':>9} {'MRR@10':>9}")
    print(f"  {pct([agg['hit@5']]):>12} {pct([agg['recall@5']]):>10} {pct([agg['recall@10']]):>10} "
          f"{pct([agg['precision@5']]):>9} {agg['mrr@10']:>9.3f}")
    print(f"  nonsense clean {agg['nonsense_clean']} · superseded outranks {agg['superseded_outranks']}")
    print(f"  latency ms  p50 {agg['latency_ms_p50']:.0f} · p95 {agg['latency_ms_p95']:.0f} "
          f"· max {agg['latency_ms_max']:.0f}")
    print(f"\n  {'category':14} {'n':>3} {'hit@5':>8} {'recall@5':>9} {'recall@10':>10} {'MRR@10':>8}")
    for c, e in agg["by_category"].items():
        if c == "nonsense":
            print(f"  {c:14} {e['n']:>3} {'clean ' + str(e['clean']) + '/' + str(e['n']):>8}"
                  f"{'':>9} {'':>10} {'avg FP ' + format(e['avg_false_positives'], '.1f'):>8}")
        else:
            extra = f"  outranks {e['outranks']}/{e['n']}" if "outranks" in e else ""
            print(f"  {c:14} {e['n']:>3} {pct([e['hit@5']]):>8} {pct([e['recall@5']]):>9} "
                  f"{pct([e['recall@10']]):>10} {e['mrr@10']:>8.3f}{extra}")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--soul", required=True)
    ap.add_argument("--corpus", required=True)
    ap.add_argument("--label", required=True)
    ap.add_argument("--gold", default=os.path.join(HERE, "gold_set.json"))
    ap.add_argument("--port", type=int, default=7893)
    ap.add_argument("--limit", type=int, default=10)
    ap.add_argument("--out", default=None)
    args = ap.parse_args()

    with open(args.gold, encoding="utf-8") as fh:
        gold = json.load(fh)
    print(f"[{args.label}] soul={os.path.basename(args.soul)} "
          f"corpus={os.path.basename(args.corpus)} gold={len(gold['queries'])}q limit={args.limit}")

    soul = Soul(args.soul, args.corpus, args.port)
    rows = []
    started = time.time()
    try:
        soul.start()
        rows = evaluate(soul, gold, args.limit)
    finally:
        dead = soul.stop()

    agg = aggregate(rows)
    print_table(args.label, agg)

    out = args.out or os.path.join(HERE, f"results-{args.label}.json")
    doc = {
        "label": args.label,
        "soul_binary": os.path.abspath(args.soul),
        "soul_md5": subprocess.run(["md5", "-q", args.soul], capture_output=True,
                                   text=True).stdout.strip(),
        "corpus": os.path.abspath(args.corpus),
        "corpus_nodes": gold.get("corpus_nodes"),
        "corpus_edges": gold.get("corpus_edges"),
        "gold_set": os.path.abspath(args.gold),
        "limit": args.limit,
        "port": args.port,
        "wall_clock_s": round(time.time() - started, 1),
        "child_pid": soul.pid,
        "child_confirmed_dead": soul.confirmed_dead,
        "aggregate": agg,
        "rows": rows,
    }
    with open(out, "w", encoding="utf-8") as fh:
        json.dump(doc, fh, indent=1, ensure_ascii=False)
    print(f"\nwrote {out}")
    if not dead:
        print("FATAL: child process could not be confirmed dead", file=sys.stderr)
        sys.exit(4)


if __name__ == "__main__":
    main()