tests/mcp_repeat_test.py

#!/usr/bin/env python3
"""Repeatability burn-in for the MCP server: fire the SAME brief N times.

Pulls the newest brief for a company from Supabase demo_history (status
mcp_intake — i.e. "Jack's last payload" for Tixr) and runs it through
build_demo_from_brief N times sequentially, polling each to completion.
Reports a per-run table + pass rate, exits non-zero if any run failed.

Every run is a fresh LLM blueprint draw, so N runs measure the real-world
flake rate (blueprint quirks, provider 529s, TS import flakes) — not just
whether one lucky draw works.

Usage (from either clone; MCP_ACCESS_TOKEN must be in the environment or in
the demoprep_mcp clone's .env, which is tried explicitly as documented):

    ./demoprep/bin/python tests/mcp_repeat_test.py                # 5x Tixr
    ./demoprep/bin/python tests/mcp_repeat_test.py --count 3
    ./demoprep/bin/python tests/mcp_repeat_test.py --company Tixr --owner jack.rayner@thoughtspot.com
"""
import argparse
import json
import os
import sys
import time

ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
sys.path.insert(0, ROOT)

from dotenv import load_dotenv

load_dotenv(os.path.join(ROOT, ".env"))
# The MCP access token historically lives in the demoprep_mcp clone's .env.
# Load it as a documented, explicit second source (does not override existing env).
_MCP_CLONE_ENV = "/Users/mike.boone/_source/demoprep_mcp/.env"
if not os.getenv("MCP_ACCESS_TOKEN") and os.path.exists(_MCP_CLONE_ENV):
    load_dotenv(_MCP_CLONE_ENV)

import tests.e2e_mcp as H  # noqa: E402
from supabase_client import SupabaseSettings  # noqa: E402


def log(msg: str) -> None:
    print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)


def fetch_latest_brief(company: str, attempts: int = 5) -> tuple[str, str, str]:
    rows = None
    for attempt in range(1, attempts + 1):
        try:
            rows = (
                SupabaseSettings().client.table("demo_history")
                .select("company_name,use_case,results,created_at")
                .eq("status", "mcp_intake")
                .ilike("company_name", f"%{company}%")
                .order("created_at", desc=True)
                .limit(1)
                .execute()
                .data
                or []
            )
            break
        except Exception as e:
            # Venue WiFi / proxy flaps surface here as SSL or connect errors;
            # retry rather than dying before the first build is even fired.
            if attempt == attempts:
                raise SystemExit(
                    f"Supabase unreachable after {attempts} attempts ({type(e).__name__}: "
                    f"{str(e)[:160]}). Check network/proxy, or pass --brief-file."
                )
            log(f"brief fetch failed ({type(e).__name__}), retry {attempt}/{attempts - 1} in 10s")
            time.sleep(10)
    if not rows:
        raise SystemExit(f"No mcp_intake brief for {company!r} in demo_history")
    row = rows[0]
    brief = (row.get("results") or {}).get("brief") or ""
    if not brief:
        raise SystemExit(f"Brief for {company!r} ({row['created_at']}) is empty")
    return brief, row.get("use_case") or "", row["created_at"][:19]


def run_once(ep: str, tok: str, brief: str, company: str, use_case: str,
             owner: str, run_no: int, timeout_min: int) -> dict:
    run_id = None
    for _ in range(60):
        try:
            resp = H.call(ep, tok, "build_demo_from_brief",
                          {"brief": brief, "company_name": company,
                           "use_case": use_case, "owner_email": owner})
        except Exception as e:
            log(f"run {run_no}: start call failed ({type(e).__name__}) — retry in 30s")
            time.sleep(30)
            continue
        status = resp.get("status")
        if status == "started":
            run_id = resp["run_id"]
            log(f"run {run_no}: started run_id={run_id}")
            break
        if status == "busy":
            log(f"run {run_no}: server busy — retry in 30s")
            time.sleep(30)
            continue
        return {"run_no": run_no, "run_id": None, "grade": "FAIL",
                "status": f"start failed: {str(resp)[:200]}", "elapsed_s": 0, "errors": []}
    if not run_id:
        return {"run_no": run_no, "run_id": None, "grade": "FAIL",
                "status": "never got a slot", "elapsed_s": 0, "errors": []}

    deadline = time.time() + timeout_min * 60
    misses = 0
    last_phase = ""
    while time.time() < deadline:
        try:
            st = H.call(ep, tok, "status")
            misses = 0
        except Exception as e:
            misses += 1
            if misses >= 20:
                return {"run_no": run_no, "run_id": run_id, "grade": "FAIL",
                        "status": "status endpoint unreachable", "elapsed_s": None, "errors": []}
            time.sleep(30)
            continue
        recent = {x["run_id"]: x for x in st.get("recent_builds", [])}
        if run_id in recent:
            res = recent[run_id]
            graded = H.grade(res)
            return {"run_no": run_no, "run_id": run_id, "grade": graded["grade"],
                    "status": res.get("status"), "elapsed_s": res.get("elapsed_seconds"),
                    "errors": res.get("errors") or [],
                    "liveboard": res.get("liveboard_url"), "schema": res.get("schema")}
        active = {x["run_id"]: x for x in st.get("active_builds", [])}.get(run_id, {})
        phase = active.get("phase", "(waiting)")
        if phase != last_phase:
            log(f"run {run_no}: {phase} {active.get('elapsed_seconds', '')}s")
            last_phase = phase
        time.sleep(25)
    return {"run_no": run_no, "run_id": run_id, "grade": "FAIL",
            "status": f"timed out after {timeout_min} min", "elapsed_s": None, "errors": []}


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--count", type=int, default=5)
    ap.add_argument("--company", default="Tixr")
    ap.add_argument("--owner", default="jack.rayner@thoughtspot.com")
    ap.add_argument("--timeout-min", type=int, default=30, help="per-run timeout")
    ap.add_argument("--brief-file", help="read the brief from a local file instead of Supabase "
                                         "(use when the network can't reach Supabase)")
    args = ap.parse_args()

    ep = os.getenv("MCP_ENDPOINT", H.DEFAULT_ENDPOINT)
    tok = os.getenv("MCP_ACCESS_TOKEN")
    if not tok:
        raise SystemExit("MCP_ACCESS_TOKEN not set (env or demoprep_mcp/.env)")

    if args.brief_file:
        brief = open(args.brief_file).read()
        use_case, brief_ts = "Event Ticketing Analytics", f"file:{args.brief_file}"
    else:
        brief, use_case, brief_ts = fetch_latest_brief(args.company)
    log(f"{args.company} brief from {brief_ts} | use_case={use_case!r} | len={len(brief)}")
    log(f"Running {args.count}x sequentially against {ep}")

    results = []
    for i in range(1, args.count + 1):
        log(f"===== RUN {i}/{args.count} =====")
        results.append(run_once(ep, tok, brief, args.company, use_case,
                                args.owner, i, args.timeout_min))

    passes = sum(1 for r in results if r["grade"] == "PASS")
    print("\n" + "=" * 78)
    print(f"{'run':>4} {'run_id':14} {'grade':6} {'status':22} {'elapsed':>9}  errors")
    print("-" * 78)
    for r in results:
        el = f"{r['elapsed_s']:.0f}s" if isinstance(r.get("elapsed_s"), (int, float)) else "-"
        err = (r["errors"][0][:60] + "…") if r.get("errors") else ""
        print(f"{r['run_no']:>4} {str(r.get('run_id'))[:12]:14} {r['grade']:6} "
              f"{str(r.get('status'))[:22]:22} {el:>9}  {err}")
    print("-" * 78)
    print(f"PASS RATE: {passes}/{len(results)}")
    print("=" * 78)
    print(json.dumps(results, indent=2, default=str))
    sys.exit(0 if passes == len(results) else 1)


if __name__ == "__main__":
    main()