test(mcp): repeatability burn-in — fire the latest brief N times, report pass rate

tests/mcp_repeat_test.py

@@ -0,0 +1,166 @@
+#!/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) -> tuple[str, str, str]:
+    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 []
+    )
+    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")
+    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)")
+
+    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()