chorus/eval/test_reflect.py

#!/usr/bin/env python3
"""test_reflect.py — H5: session-reflection capture (dry-run vs --apply).

A dry-run previews and writes NOTHING; --apply routes each proposal through capture
(creating notes, the inbox line, and skipping low-confidence items). Drives the real
chorus.py against eval/mock_olrapi.py. No creds, no live vault.

Run:  python test_reflect.py [--port 8850]
"""
from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
import tempfile
import time
import urllib.request
from pathlib import Path

HERE = Path(__file__).resolve().parent
CHORUS = HERE.parent / "chorus-memory.plugin.src" / "skills" / "chorus-memory" / "scripts" / "chorus.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)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--port", type=int, default=8850)
    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)

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

    def chorus(*args, stdin=None):
        env = dict(os.environ, CHORUS_BASE=base, CHORUS_KEY=KEY, CHORUS_VERIFY="1", CHORUS_TODAY="2026-06-22")
        return subprocess.run([sys.executable, str(CHORUS), *args], input=stdin,
                              capture_output=True, text=True, env=env)

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

    try:
        for _ in range(50):
            try:
                urllib.request.urlopen(f"{base}/__debug__reset", data=b"", timeout=1); break
            except Exception:
                time.sleep(0.1)
        http("PUT", f"{base}/vault/_agent/chorus-vault.md", "---\nschema_version: 4\n---\n# marker\n")

        proposals = [
            {"title": "Acme Corp", "kind": "company", "body": "A vendor CHORUS integrates with.", "confidence": 0.9},
            {"title": "Use uv not pip", "kind": "semantic", "body": "Jason standardizes on uv.", "confidence": 0.95},
            {"title": "half-formed idea", "inbox": True, "confidence": 0.9},
            {"title": "Maybe relevant", "kind": "concept", "confidence": 0.2},  # below floor -> skipped
        ]
        pfile = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False, encoding="utf-8")
        json.dump(proposals, pfile); pfile.close()

        # dry-run: previews, writes nothing
        r = chorus("reflect", pfile.name)
        check("dry-run previews", "dry-run" in r.stdout and "Acme Corp" in r.stdout, r.stdout)
        check("dry-run drops the low-confidence proposal", "Maybe relevant" not in r.stdout, r.stdout)
        check("dry-run writes nothing", ground("resources/companies/acme-corp.md") is None)

        # --apply: routes each through capture
        r = chorus("reflect", pfile.name, "--apply")
        check("apply reports applied count", "applied" in r.stdout, r.stdout + r.stderr)
        check("apply creates the company note",
              (ground("resources/companies/acme-corp.md") or "").find("type: company") >= 0)
        check("apply creates the semantic note",
              ground("_agent/memory/semantic/use-uv-not-pip.md") is not None)
        check("apply routes the inbox proposal", "half-formed idea" in (ground("inbox/captures/inbox.md") or ""))
        check("apply still skips the low-confidence proposal",
              ground("resources/concepts/maybe-relevant.md") is None)

        # stdin path also works (proposals piped, not a file)
        r = chorus("reflect", "-", stdin=json.dumps([{"title": "Piped Co", "kind": "company", "confidence": 0.9}]))
        check("reflect reads proposals from stdin", "Piped Co" in r.stdout, r.stdout + r.stderr)

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


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