#!/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()