chorus/eval/test_features.py

#!/usr/bin/env python3
"""test_features.py — end-to-end test of the v0.9 features against the mock OLRAPI.

Exercises the real shipped scripts (chorus.py capture/resolve/recall/link + sweep.py)
through subprocess, then reads ground truth back from the mock. No network, no creds,
no live vault.

Run:  python test_features.py [--port 8801]
"""
import argparse, json, os, subprocess, sys, time, urllib.request, urllib.error
from pathlib import Path

HERE = Path(__file__).resolve().parent
SCRIPTS = HERE.parent / "chorus-memory.plugin.src" / "skills" / "chorus-memory" / "scripts"
CHORUS = SCRIPTS / "chorus.py"
SWEEP = SCRIPTS / "sweep.py"
KEY = "test-key-not-a-real-secret"

failures = []


def check(name, cond, detail=""):
    print(f"{'ok  ' if cond else 'FAIL'} {name}" + (f"  -- {detail}" if not cond else ""))
    if not cond:
        failures.append(name)


class Harness:
    def __init__(self, base):
        self.base = base

    def http(self, method, url, body=None, headers=None):
        data = body.encode() if isinstance(body, str) else body
        req = urllib.request.Request(url, data=data, method=method,
                                     headers={"Authorization": f"Bearer {KEY}", **(headers or {})})
        try:
            with urllib.request.urlopen(req, timeout=10) as r:
                return r.status, r.read().decode("utf-8", "replace")
        except urllib.error.HTTPError as e:
            return e.code, e.read().decode("utf-8", "replace")
        except Exception as e:
            return 0, str(e)

    def reset(self):
        self.http("POST", f"{self.base}/__debug__reset")

    def seed(self, path, content):
        self.http("PUT", f"{self.base}/vault/{path}", content)

    def ground(self, path):
        _, body = self.http("GET", f"{self.base}/__debug__?path={path}")
        return None if body == "<<MISSING>>" else body

    def chorus(self, *args):
        env = dict(os.environ, CHORUS_BASE=self.base, CHORUS_KEY=KEY, CHORUS_MEMBER="eval-member", CHORUS_VERIFY="1",
                   CHORUS_TODAY="2026-06-21")
        return subprocess.run([sys.executable, str(args[0]), *args[1:]],
                              capture_output=True, text=True, env=env)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--port", type=int, default=8801)
    a = ap.parse_args()
    base = f"http://127.0.0.1:{a.port}"
    srv = subprocess.Popen([sys.executable, str(HERE / "mock_olrapi.py"), "--port", str(a.port)],
                           stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
    try:
        for _ in range(50):
            try:
                urllib.request.urlopen(f"{base}/__debug__reset", data=b"", timeout=1); break
            except Exception:
                time.sleep(0.1)
        h = Harness(base)
        h.reset()
        h.seed("_agent/chorus-vault.md", "---\nschema_version: 2\n---\n# marker\n")

        # 1. capture a person (with an alias)
        r = h.chorus(CHORUS, "capture", "Bob Smith", "--kind", "person", "--aliases", "bob,rs")
        note = h.ground("resources/people/bob-smith.md")
        check("capture person creates note", note is not None and "type: person" in note, r.stderr)
        check("capture stamps agent_written", note and "agent_written: true" in note)
        check("capture stamps author with configured member",
              note and "author: eval-member" in note)
        idx = h.ground("_agent/index/entities.json")
        check("index records entity", idx and "bob-smith" in idx and '"bob"' in idx)

        # 2. resolve by alias
        r = h.chorus(CHORUS, "resolve", "bob")
        out = json.loads(r.stdout) if r.stdout.strip().startswith("{") else {}
        check("resolve alias -> path", out.get("match") and out.get("path") == "resources/people/bob-smith.md", r.stdout)

        # 3. capture a company whose body mentions Bob Smith -> auto bidirectional link
        r = subprocess.run([sys.executable, str(CHORUS), "capture", "MPM", "--kind", "company", "-"],
                           input="Bob Smith is the principal here.\n", capture_output=True, text=True,
                           env=dict(os.environ, CHORUS_BASE=base, CHORUS_KEY=KEY, CHORUS_MEMBER="eval-member", CHORUS_TODAY="2026-06-21"))
        mpm = h.ground("resources/companies/mpm.md")
        bob = h.ground("resources/people/bob-smith.md")
        check("auto-link forward (mpm -> bob)", mpm and "[[resources/people/bob-smith]]" in mpm, r.stdout + r.stderr)
        check("auto-link reciprocal (bob -> mpm)", bob and "[[resources/companies/mpm]]" in bob)

        # 4. recall an entity surfaces its linked neighborhood (search is mocked empty,
        #    so this exercises the index-resolve + 1-hop expansion path)
        r = h.chorus(CHORUS, "recall", "MPM")
        check("recall surfaces linked person", "resources/people/bob-smith.md" in r.stdout, r.stdout)

        # 4b. BM25 finds a note by a BODY term. /search/simple is mocked empty, so a hit
        #     here can only come from the local BM25 recall index (proves H1 lexical layer).
        r = h.chorus(CHORUS, "recall", "principal")
        check("recall finds note by body term (BM25)",
              "resources/companies/mpm.md" in r.stdout, r.stdout)

        # 5. explicit link between two fresh notes
        h.chorus(CHORUS, "capture", "Alpha", "--kind", "concept")
        h.chorus(CHORUS, "capture", "Beta", "--kind", "concept")
        h.chorus(CHORUS, "link", "resources/concepts/alpha.md", "resources/concepts/beta.md")
        alpha = h.ground("resources/concepts/alpha.md")
        beta = h.ground("resources/concepts/beta.md")
        check("link adds A->B", alpha and "[[resources/concepts/beta]]" in alpha)
        check("link adds B->A", beta and "[[resources/concepts/alpha]]" in beta)

        # 6. inbox capture (unknown home)
        h.chorus(CHORUS, "capture", "stray thought", "--inbox")
        inbox = h.ground("_agent/members/eval-member/inbox.md")
        check("inbox capture lands", inbox and "stray thought" in inbox)

        # 7. sweep --apply rebuilds the index, symmetrizes, stamps schema 3
        r = h.chorus(SWEEP, "--apply")
        marker = h.ground("_agent/chorus-vault.md")
        check("sweep runs clean", r.returncode == 0, r.stdout + r.stderr)
        check("sweep stamps schema 5", marker and "schema_version=5" in marker.replace(": ", "="), marker)
        idx2 = h.ground("_agent/index/entities.json")
        check("sweep rebuilt index has all entities",
              idx2 and all(s in idx2 for s in ["bob-smith", "mpm", "alpha", "beta"]))
        rix = h.ground("_agent/index/recall-index.json")
        check("sweep builds the recall (BM25) index",
              rix is not None and '"postings"' in rix and "principal" in rix, rix)

        # 8. H3 — atomic_index_update re-reads fresh and MERGES, so an entry written by a
        #    "concurrent" session survives our write; and the advisory lock is released after.
        h.http("PUT", f"{base}/vault/_agent/index/entities.json",
               '{"schema":1,"updated":"2026-06-21","entities":{"ext-entity":'
               '{"path":"resources/concepts/ext-entity.md","kind":"concept","title":"Ext",'
               '"aliases":[],"last_seen":"2026-06-21"}}}')
        env = dict(os.environ, CHORUS_BASE=base, CHORUS_KEY=KEY, CHORUS_MEMBER="eval-member", CHORUS_TODAY="2026-06-21", PYTHONPATH=str(SCRIPTS))
        snippet = ("import chorus_index as ix, chorus_concurrency as c; "
                   "c.atomic_index_update(lambda d: ix.upsert(d,'mine',"
                   "'resources/concepts/mine.md','concept','Mine'))")
        subprocess.run([sys.executable, "-c", snippet], env=env, capture_output=True, text=True)
        idx3 = h.ground("_agent/index/entities.json")
        check("H3 atomic update merges a concurrent entry (no clobber)",
              bool(idx3) and "ext-entity" in idx3 and "mine" in idx3, idx3)
        check("H3 advisory lock released after the transaction",
              h.ground("_agent/locks/vault.lock") is None)

        # 9. M4 — --dry-run previews and writes nothing; --json emits a parseable envelope.
        r = h.chorus(CHORUS, "capture", "DryRun Co", "--kind", "company", "--dry-run", "--json")
        try:
            env = json.loads(r.stdout)
        except Exception:
            env = {}
        check("M4 --dry-run emits a dry-run JSON envelope",
              env.get("action", "").startswith("dry-run") and env.get("path") == "resources/companies/dryrun-co.md", r.stdout)
        check("M4 --dry-run writes nothing", h.ground("resources/companies/dryrun-co.md") is None)
        r = h.chorus(CHORUS, "capture", "Json Co", "--kind", "company", "--json")
        try:
            env = json.loads(r.stdout)
        except Exception:
            env = {}
        check("M4 --json envelope reports created + path",
              env.get("ok") is True and env.get("action") == "created"
              and env.get("path") == "resources/companies/json-co.md", r.stdout)
        check("M4 --json capture actually wrote the note", h.ground("resources/companies/json-co.md") is not None)

        # ---------------- v1.5 features ----------------------------------------
        # step 8 clobbered entities.json on purpose; rebuild the index from the notes
        # so the v1.5 tests run against a coherent vault.
        h.chorus(SWEEP, "--apply")

        # 10. frontmatter completeness at write time: entity notes are born classified.
        bob2 = h.ground("resources/people/bob-smith.md") or ""
        check("v1.5 capture stamps a default status", "status: active" in bob2, bob2)
        check("v1.5 capture seeds tags with the kind", "tags: [person]" in bob2, bob2)

        # 11. update-capture preserves the FULL body (old code kept only line 1).
        r = subprocess.run([sys.executable, str(CHORUS), "capture", "Bob Smith", "-", "--kind", "person"],
                           input="met at expo\nsecond line survives\nthird line too\n",
                           capture_output=True, text=True,
                           env=dict(os.environ, CHORUS_BASE=base, CHORUS_KEY=KEY, CHORUS_MEMBER="eval-member", CHORUS_TODAY="2026-06-21"))
        bob3 = h.ground("resources/people/bob-smith.md") or ""
        check("v1.5 update keeps the dated bullet", "- 2026-06-21 [eval-member]: met at expo" in bob3, r.stdout + r.stderr)
        check("v1.5 update keeps EVERY body line",
              "  second line survives" in bob3 and "  third line too" in bob3, bob3)

        # 12. duplicate gate: same-kind + distinctive-token candidate STOPS creation...
        r = h.chorus(CHORUS, "capture", "Robert Smith", "--kind", "person")
        check("v1.5 duplicate gate stops the create (exit 76)", r.returncode == 76, str(r.returncode))
        check("v1.5 gate wrote nothing", h.ground("resources/people/robert-smith.md") is None)
        # ... --force overrides ...
        r = h.chorus(CHORUS, "capture", "Robert Smith", "--kind", "person", "--force")
        check("v1.5 --force creates past the gate",
              h.ground("resources/people/robert-smith.md") is not None, r.stdout + r.stderr)
        # 12b (1.5.1). cross-kind lookalike does NOT gate — it creates with a warning.
        r = h.chorus(CHORUS, "capture", "Smith Consulting", "--kind", "company")
        check("v1.5.1 cross-kind lookalike is not gated",
              r.returncode == 0 and h.ground("resources/companies/smith-consulting.md") is not None,
              str(r.returncode) + r.stdout + r.stderr)
        check("v1.5.1 cross-kind lookalike still warns",
              "similar existing" in (r.stdout + r.stderr), r.stdout + r.stderr)
        # 12c (1.5.1). a single vault-common token does NOT gate: once "acme" appears in
        # two entities, "Acme Tools" creates (with a warning) instead of blocking.
        h.chorus(CHORUS, "capture", "Acme", "--kind", "company")
        h.chorus(CHORUS, "capture", "Acme Group", "--kind", "company", "--force")
        r = h.chorus(CHORUS, "capture", "Acme Tools", "--kind", "company")
        check("v1.5.1 single common token does not gate",
              r.returncode == 0 and h.ground("resources/companies/acme-tools.md") is not None,
              str(r.returncode) + r.stdout + r.stderr)
        # ... and --merge-into routes the capture as an update to the canonical entity.
        r = h.chorus(CHORUS, "capture", "Bobby The Builder", "--kind", "person", "--merge-into", "bob-smith")
        idx4 = h.ground("_agent/index/entities.json") or ""
        check("v1.5 --merge-into updates the existing entity (mention learned as alias)",
              r.returncode == 0 and "Bobby The Builder" in idx4, r.stdout + r.stderr)
        check("v1.5 --merge-into did not spawn a new note",
              h.ground("resources/people/bobby-the-builder.md") is None)

        # 13. recall corpus now includes session logs (down-weighted, but findable).
        h.seed("_agent/sessions/2026-06-15-1200-mpm-review.md",
               "---\ntype: session-log\ncreated: 2026-06-15\nupdated: 2026-06-15\n---\n"
               "# MPM review\n\nDiscussed the kerfuffle about invoicing.\n")
        h.chorus(SWEEP, "--apply")
        r = h.chorus(CHORUS, "recall", "kerfuffle")
        check("v1.5 recall finds a session log by body term",
              "_agent/sessions/2026-06-15-1200-mpm-review.md" in r.stdout, r.stdout)
        rix2 = h.ground("_agent/index/recall-index.json") or ""
        check("v1.5 recall index carries doc meta (schema 2)",
              '"schema": 2' in rix2 or '"schema":2' in rix2, rix2[:200])

        # 14. recency-aware ranking: same term, fresher note wins.
        h.seed("resources/concepts/old-idea.md",
               "---\ntype: concept\nstatus: active\ncreated: 2025-01-01\nupdated: 2025-01-01\n"
               "tags: [concept]\n---\n# Old Idea\n\nzeppelin research notes\n")
        h.seed("resources/concepts/new-idea.md",
               "---\ntype: concept\nstatus: active\ncreated: 2026-06-20\nupdated: 2026-06-20\n"
               "tags: [concept]\n---\n# New Idea\n\nzeppelin research notes\n")
        h.chorus(SWEEP, "--apply")
        r = h.chorus(CHORUS, "recall", "zeppelin", "--json")
        try:
            env2 = json.loads(r.stdout)
        except Exception:
            env2 = {}
        primary = env2.get("primary") or []
        check("v1.5 recall --json emits structured hits",
              env2.get("ok") is True and len(primary) >= 2, r.stdout[:300])
        check("v1.5 fresher note outranks stale twin",
              primary and primary[0].get("path") == "resources/concepts/new-idea.md",
              json.dumps(primary[:2]))
        check("v1.5 recall hits carry updated/status metadata",
              primary and primary[0].get("updated") == "2026-06-20"
              and primary[0].get("status") == "active", json.dumps(primary[:1]))

        # 15. one-tap triage: structured list, then route with an audit trail.
        h.seed("_agent/members/eval-member/inbox.md",
               "- 2026-06-01: prefers uv over pip\n- 2026-06-20: try the new espresso place\n")
        r = h.chorus(CHORUS, "triage", "--list", "--json")
        try:
            tenv = json.loads(r.stdout)
        except Exception:
            tenv = {}
        items = tenv.get("items") or []
        check("v1.5 triage --list parses dated captures with ages",
              len(items) == 2 and items[0].get("age_days") == 20, r.stdout[:300])
        props = [{"title": "Prefers uv over pip", "kind": "semantic",
                  "body": "The operator prefers uv over pip for Python tooling.",
                  "line": "- 2026-06-01: prefers uv over pip", "confidence": 0.9}]
        pfile = HERE / "_triage_props.json"
        pfile.write_text(json.dumps(props), encoding="utf-8")
        r = h.chorus(CHORUS, "triage", str(pfile), "--apply")
        pfile.unlink()
        routed = h.ground("_agent/memory/semantic/prefers-uv-over-pip.md")
        plog = h.ground("inbox/processing-log/2026-06-21.md") or ""
        check("v1.5 triage routes the item via capture", routed is not None, r.stdout + r.stderr)
        check("v1.5 triage writes the processing-log audit line",
              "prefers uv over pip" in plog and "_agent/memory/semantic/prefers-uv-over-pip.md" in plog, plog)
        check("v1.5 triage keeps the original capture",
              "prefers uv over pip" in (h.ground("_agent/members/eval-member/inbox.md") or ""))

        # 16. lint flags incomplete entity frontmatter; sweep backfills it.
        h.seed("resources/companies/driftco.md",
               "---\ntype: company\ncreated: 2026-06-01\nupdated: 2026-06-01\ntags: []\n---\n# DriftCo\n")
        LINT = SCRIPTS / "vault_lint.py"
        r = h.chorus(LINT)
        check("v1.5 lint flags missing status", "driftco.md: missing status" in r.stdout, r.stdout[-800:])
        check("v1.5 lint flags empty tags", "driftco.md: empty tags" in r.stdout, r.stdout[-800:])
        h.chorus(SWEEP, "--apply")
        drift = h.ground("resources/companies/driftco.md") or ""
        # (the naive mock records frontmatter PATCHes as <fm:...> markers instead of
        # editing the YAML; the hi-fi suite proves the real fm semantics)
        check("v1.5 sweep backfills status",
              "status: active" in drift or '<fm:status="active">' in drift, drift)
        check("v1.5 sweep backfills the kind tag",
              "tags: [company]" in drift or '<fm:tags=["company"]>' in drift, drift)

        # 17. hooks: stop-hook nudges once on a substantive session, then stays quiet.
        import tempfile
        tdir = Path(tempfile.mkdtemp())
        transcript = tdir / "t.jsonl"
        turns = [json.dumps({"type": "user", "message": {"content": f"user turn {i}"}}) for i in range(6)]
        transcript.write_text("\n".join(turns), encoding="utf-8")
        hook_env = dict(os.environ, CHORUS_BASE=base, CHORUS_KEY=KEY, CHORUS_MEMBER="eval-member", CHORUS_STATE_DIR=str(tdir))
        payload = json.dumps({"session_id": "s1", "transcript_path": str(transcript),
                              "stop_hook_active": False})
        HOOK_STOP = SCRIPTS / "chorus_hook_stop.py"
        r = subprocess.run([sys.executable, str(HOOK_STOP)], input=payload,
                           capture_output=True, text=True, env=hook_env)
        try:
            henv = json.loads(r.stdout)
        except Exception:
            henv = {}
        check("v1.5 stop hook nudges a substantive session",
              r.returncode == 0 and henv.get("decision") == "block", r.stdout + r.stderr)
        r = subprocess.run([sys.executable, str(HOOK_STOP)], input=payload,
                           capture_output=True, text=True, env=hook_env)
        check("v1.5 stop hook fires only once per session",
              r.returncode == 0 and not r.stdout.strip(), r.stdout)
        HOOK_START = SCRIPTS / "chorus_hook_session_start.py"
        r = subprocess.run([sys.executable, str(HOOK_START)],
                           input=json.dumps({"source": "startup"}),
                           capture_output=True, text=True, env=hook_env)
        try:
            senv = json.loads(r.stdout)
        except Exception:
            senv = {}
        ctx = (senv.get("hookSpecificOutput") or {}).get("additionalContext", "")
        check("v1.5 session-start hook injects the load output",
              r.returncode == 0 and "marker" in ctx and "chorus-vault.md" in ctx, r.stdout[:300])
        r = subprocess.run([sys.executable, str(HOOK_START)],
                           input=json.dumps({"source": "resume"}),
                           capture_output=True, text=True, env=hook_env)
        check("v1.5 session-start hook stays quiet on resume",
              r.returncode == 0 and not r.stdout.strip(), r.stdout)

        print(f"\n{len(failures)} failure(s)" if failures else "\nall feature tests passed")
        return 1 if failures else 0
    finally:
        srv.terminate()


if __name__ == "__main__":
    raise SystemExit(main())