fix: identify e2e runs from final model URL

tests/e2e_quality.py

@@ -437,8 +437,8 @@ def submit_job(page: Page, test_case: dict):
     url_el.fill(test_case["company_url"])
     page.wait_for_timeout(300)
 
-    # Generate a unique tag for this test case — logged by the app at session init
-    # so fetch_run_diagnostics can find this session without any guessing
+    # Generate a unique tag for this test case. Logs are diagnostic-only; the
+    # completed page's model/liveboard URLs identify the run under test.
     test_tag = f"TR-{uuid.uuid4().hex[:8].upper()}"
     test_case["_test_tag"] = test_tag  # store so run_single_test can pass it to diagnostics
 
@@ -504,17 +504,28 @@ def pipeline_finished(stages: dict) -> bool:
 def extract_run_context(page: Page) -> dict:
     """
     After completion, find model and liveboard URLs in the page.
-    Uses full HTML content so GUIDs in href attributes are also matched.
+    Uses visible text plus full HTML so GUIDs in href attributes are also
+    matched. This is the source of truth for the run under test.
     """
-    body = page.content()  # full HTML — catches GUIDs in href attrs too
+    try:
+        visible = page.inner_text("body", timeout=5000)
+    except Exception:
+        visible = ""
+    body = f"{visible}\n{page.content()}"  # full HTML catches GUIDs in href attrs too
 
     model_match = re.search(r'(https://[^\s"\'<>#]+)/#/data/tables/([a-f0-9-]{36})', body)
     lb_match    = re.search(r'(https://[^\s"\'<>#]+)/#/pinboard/([a-f0-9-]{36})',    body)
+    ts_base = None
+    if model_match:
+        ts_base = model_match.group(1)
+    elif lb_match:
+        ts_base = lb_match.group(1)
 
     return {
-        "ts_base_url":    model_match.group(1) if model_match else None,
+        "ts_base_url":    ts_base,
         "model_guid":     model_match.group(2) if model_match else None,
         "liveboard_guid": lb_match.group(2)    if lb_match    else None,
+        "source":          "page",
     }
 
 
@@ -610,6 +621,26 @@ def extract_db_schema(model_tml_str: str) -> tuple:
     return "", ""
 
 
+def resolve_schema_from_model(run_context: dict) -> dict:
+    """
+    Resolve the Snowflake schema from the exact ThoughtSpot model printed by
+    the app. No prefix/date guessing.
+    """
+    ts_base = run_context.get("ts_base_url")
+    model_guid = run_context.get("model_guid")
+    if not ts_base or not model_guid:
+        return {"found": False, "reason": "missing model URL"}
+    try:
+        session = ts_authenticate(ts_base)
+        model_tml = export_model_tml(ts_base, session, model_guid)
+        db, schema = extract_db_schema(model_tml)
+        if not schema:
+            return {"found": False, "reason": "model TML did not expose schema"}
+        return {"found": True, "database": db or "DEMOBUILD", "schema": schema}
+    except Exception as e:
+        return {"found": False, "reason": str(e)}
+
+
 def get_snowflake_sample(db: str, schema: str) -> str:
     try:
         from snowflake_auth import get_snowflake_connection
@@ -804,7 +835,7 @@ def run_ai_grading(run_context: dict, company: str, vertical: str, line: str, fu
         db, schema = extract_db_schema(model_tml)
         if (not db or not schema) and schema_override:
             db, schema = "DEMOBUILD", schema_override
-            print(f"    ℹ️  Using schema from session_logs: {schema_override}")
+            print(f"    ℹ️  Using schema resolved from model: {schema_override}")
         sample = get_snowflake_sample(db, schema) if db and schema else "Could not determine db/schema"
         print("    🤖 Grading data quality...")
         dg    = grade_data_quality(company, vertical, line, function, model_tml, sample)
@@ -1140,8 +1171,8 @@ def compute_grade(score: float, config: dict) -> str:
 def fetch_run_diagnostics(start_time: float, company: str = "", test_tag: str = "") -> dict:
     """
     Query session_logs for entries by testrunner after start_time.
-    If test_tag is provided (preferred), matches the session by exact tag lookup.
-    Falls back to company prefix matching when no tag is available.
+    Matches only by exact test tag. This is intentionally not used to identify
+    the model/schema under test because concurrent runs can contaminate logs.
     """
     try:
         from supabase_client import SupabaseSettings
@@ -1169,57 +1200,16 @@ def fetch_run_diagnostics(start_time: float, company: str = "", test_tag: str =
             sid = log["session_id"]
             sessions.setdefault(sid, []).append(log)
 
-        # Derive expected 3-letter company prefix from company name/URL
-        # e.g. "delta.com" → "DEL", "Wells Fargo" → "WEL"
-        company_prefix = ""
-        if company:
-            name = company.lower().replace(".com", "").replace(".", "").replace(" ", "")
-            company_prefix = name[:3].upper()
-
-        def schema_for_session(session_entries):
-            for l in session_entries:
-                schema = (l.get("meta") or {}).get("schema", "")
-                if schema:
-                    return schema
-            return ""
-
-        # 1. Exact tag match — most reliable, no guessing
         if test_tag:
             tag_matching = {
                 sid: entries for sid, entries in sessions.items()
                 if any((l.get("meta") or {}).get("test_tag") == test_tag for l in entries)
             }
-            if tag_matching:
-                session_logs = max(tag_matching.values(), key=len)
-            else:
-                # Tag not found yet (pipeline may still be starting) — fall through to prefix
-                if company_prefix:
-                    matching = {
-                        sid: entries for sid, entries in sessions.items()
-                        if company_prefix in schema_for_session(entries).upper()
-                    }
-                    if matching:
-                        session_logs = max(matching.values(), key=len)
-                    else:
-                        session_logs = max(sessions.values(), key=len)
-                        schema = schema_for_session(session_logs)
-                        print(f"    ⚠️  Tag '{test_tag}' not found, prefix '{company_prefix}' no match — using best guess (schema: {schema})")
-                else:
-                    session_logs = max(sessions.values(), key=len)
-        # 2. Company prefix match — fallback when no tag
-        elif company_prefix:
-            matching = {
-                sid: entries for sid, entries in sessions.items()
-                if company_prefix in schema_for_session(entries).upper()
-            }
-            if matching:
-                session_logs = max(matching.values(), key=len)
-            else:
-                session_logs = max(sessions.values(), key=len)
-                schema = schema_for_session(session_logs)
-                print(f"    ⚠️  No session matched company prefix '{company_prefix}' — using best guess (schema: {schema})")
+            if not tag_matching:
+                return {"found": False, "reason": f"Exact test tag not found in session_logs: {test_tag}"}
+            session_logs = max(tag_matching.values(), key=len)
         else:
-            session_logs = max(sessions.values(), key=len)
+            return {"found": False, "reason": "No test tag provided; refusing to guess session"}
 
         # Summarise
         completed  = [l["stage"] for l in session_logs if "completed" in (l.get("event") or "")]
@@ -1277,29 +1267,18 @@ def fetch_run_diagnostics(start_time: float, company: str = "", test_tag: str =
 def check_snowflake_schema(company: str, start_time: float, schema_override: str = None) -> dict:
     """
     Get row counts for the Snowflake schema created during this test run.
-    Uses schema_override (from session_logs) if available — no guessing needed.
-    Falls back to prefix+date matching only if session_logs didn't record the schema.
+    Requires an explicit schema resolved from the exact ThoughtSpot model.
+    Never guesses by company/date prefix.
     """
     try:
         from snowflake_auth import get_snowflake_connection
-        from datetime import datetime
 
+        if not schema_override:
+            return {"found": False, "reason": "No explicit schema provided; refusing to guess"}
         conn   = get_snowflake_connection()
         cursor = conn.cursor()
 
-        if schema_override:
-            schema = schema_override
-        else:
-            # Fall back: guess from company prefix + date
-            prefix   = company[:3].upper()
-            date_str = datetime.fromtimestamp(start_time).strftime("%m%d")
-            cursor.execute("SHOW SCHEMAS IN DATABASE DEMOBUILD")
-            all_schemas = [row[1] for row in cursor.fetchall()]
-            matching = [s for s in all_schemas if s.startswith(prefix) and date_str in s]
-            if not matching:
-                cursor.close(); conn.close()
-                return {"found": False, "prefix": prefix, "date": date_str}
-            schema = sorted(matching)[-1]  # most recent
+        schema = schema_override
         cursor.execute(f'SHOW TABLES IN SCHEMA DEMOBUILD."{schema}"')
         tables = cursor.fetchall()
 
@@ -1344,7 +1323,7 @@ def print_diagnostics(diag: dict, sf: dict):
         for t in sf["tables"]:
             print(f"    {t['table']}: {t['rows']} rows")
     elif sf:
-        print(f"  Snowflake: no schema found for prefix={sf.get('prefix')} date={sf.get('date')}")
+        print(f"  Snowflake: {sf.get('reason') or sf.get('error') or 'schema not found'}")
     print("  ─────────────────────────────────────────────────────")
 
 
@@ -1432,30 +1411,29 @@ def run_single_test(page: Page, test_case: dict, config: dict) -> dict:
 
     result["duration_seconds"] = round(time.time() - start)
 
-    # Always fetch session_logs — primary source for GUIDs, schema, and errors
+    # The completed page prints the exact model/liveboard URLs for this run.
+    # Treat that as authoritative; session_logs are diagnostics only.
+    page_ctx = extract_run_context(page)
+    result["run_context"] = page_ctx
+    if page_ctx.get("model_guid"):
+        lb_note = page_ctx.get("liveboard_guid", "")
+        print(f"  🔑 GUIDs from page: model={page_ctx['model_guid'][:8]}… lb={lb_note[:8] if lb_note else 'none'}…")
+    else:
+        print("  ⚠️  No model GUID found on final page — AI grading will be skipped")
+
+    # Fetch exact-tag diagnostics only for supplemental errors/stage reconciliation.
     diag = fetch_run_diagnostics(start, company=result.get("company", ""),
                                  test_tag=test_case.get("_test_tag", ""))
     result["diagnostics"] = diag
+    if not diag.get("found"):
+        print(f"  ℹ️  Session logs not used for identity: {diag.get('reason')}")
 
-    # Build run_context from session_logs (authoritative); fall back to page scrape only if missing
-    log_ctx = {
-        "ts_base_url":    diag.get("ts_base_url"),
-        "model_guid":     diag.get("model_guid"),
-        "liveboard_guid": diag.get("liveboard_guid"),
-    }
-    if log_ctx["model_guid"]:
-        result["run_context"] = log_ctx
-        print(f"  🔑 GUIDs from session_logs: model={log_ctx['model_guid'][:8]}… lb={log_ctx['liveboard_guid'] and log_ctx['liveboard_guid'][:8] or 'none'}…")
+    schema_resolution = resolve_schema_from_model(result["run_context"])
+    known_schema = schema_resolution.get("schema") if schema_resolution.get("found") else None
+    if known_schema:
+        print(f"  🧭 Schema from ThoughtSpot model: {known_schema}")
     else:
-        page_ctx = extract_run_context(page)
-        result["run_context"] = page_ctx
-        if page_ctx.get("model_guid"):
-            print(f"  🔑 GUIDs from page HTML (log miss): model={page_ctx['model_guid'][:8]}…")
-        else:
-            print("  ⚠️  No model GUID found in logs or page — AI grading will be skipped")
-
-    # Use the real schema name from session_logs; fall back to prefix-guessing only if missing
-    known_schema = diag.get("snowflake_schema") if diag.get("found") else None
+        print(f"  ⚠️  Could not resolve schema from model: {schema_resolution.get('reason')}")
     sf = check_snowflake_schema(result["company"], start, schema_override=known_schema)
     result["snowflake_check"] = sf
     if sf.get("found"):
@@ -1463,7 +1441,7 @@ def run_single_test(page: Page, test_case: dict, config: dict) -> dict:
         for t in sf["tables"]:
             print(f"    {t['table']}: {t['rows']} rows")
     else:
-        print(f"  📦 Snowflake: schema not found (logged as: {known_schema or 'not in logs'})")
+        print(f"  📦 Snowflake: schema not checked ({sf.get('reason') or sf.get('error') or 'unknown'})")
 
     # Reconcile stages with session_logs — catches cases where the monitor bailed early
     # but the pipeline actually completed (e.g. slow DDL that took >20 min)
@@ -1488,6 +1466,7 @@ def run_single_test(page: Page, test_case: dict, config: dict) -> dict:
         ag = run_ai_grading(
             result["run_context"],
             result["company"], result["vertical"], result["line"], result["function"],
+            schema_override=known_schema,
         )
     else:
         ag["grading_errors"].append("Skipped — no model GUID (pipeline did not complete)")