fix: harden ThoughtSpot table polling and model readiness

tests/test_liveboard_creation_context.py

@@ -121,6 +121,24 @@ table:
     assert parsed["table"]["connection"]["fqn"] == "conn-guid"
 
 
+def test_metadata_header_parser_accepts_search_response_shape():
+    deployer = thoughtspot_deployer.ThoughtSpotDeployer.__new__(
+        thoughtspot_deployer.ThoughtSpotDeployer
+    )
+
+    header = deployer._metadata_header(
+        {
+            "metadata_header": {
+                "name": "DATES",
+                "id_guid": "table-guid",
+            }
+        }
+    )
+
+    assert header["name"] == "DATES"
+    assert header["id_guid"] == "table-guid"
+
+
 def test_tml_504_recovery_verifies_state_before_retrying_full_payload():
     source = inspect.getsource(thoughtspot_deployer.ThoughtSpotDeployer.deploy_all)
 
@@ -128,3 +146,4 @@ def test_tml_504_recovery_verifies_state_before_retrying_full_payload():
     assert "_verify_table_updates_after_timeout" in source
     assert "retrying full {retry_kind} payload once" in source
     assert "connection_fqn=connection_fqn" in source
+    assert "wait_for_model_answer_ready" in source

thoughtspot_deployer.py

@@ -531,6 +531,102 @@ class ThoughtSpotDeployer:
             print(f"   ⚠️ Error getting model columns: {e}")
             return []
 
+    def wait_for_model_answer_ready(
+        self,
+        model_guid: str,
+        model_columns: List[Dict],
+        log_callback=None,
+        session_logger=None,
+        timeout_seconds: int = None,
+        poll_interval_seconds: int = None,
+    ) -> bool:
+        """Wait until ThoughtSpot's answer service can query the newly-created model."""
+        timeout_seconds = timeout_seconds if timeout_seconds is not None else int(
+            os.getenv("TS_MODEL_READY_TIMEOUT_SECONDS", "300")
+        )
+        poll_interval_seconds = poll_interval_seconds if poll_interval_seconds is not None else int(
+            os.getenv("TS_MODEL_READY_POLL_INTERVAL_SECONDS", "30")
+        )
+        poll_interval_seconds = max(1, poll_interval_seconds)
+        timeout_seconds = max(poll_interval_seconds, timeout_seconds)
+
+        def _log(message):
+            if log_callback:
+                log_callback(message)
+            else:
+                print(message, flush=True)
+
+        measure = next((col for col in model_columns if col.get("type") == "NUMBER"), None)
+        if measure:
+            probe_query = f"sum [{measure.get('name')}]"
+        elif model_columns:
+            probe_query = f"[{model_columns[0].get('name')}]"
+        else:
+            probe_query = None
+
+        if not probe_query:
+            _log("   ⚠️  Model readiness check skipped: no columns available")
+            return False
+
+        deadline = time.time() + timeout_seconds
+        attempt = 0
+        last_error = ""
+
+        while True:
+            attempt += 1
+            try:
+                response = self.session.post(
+                    f"{self.base_url}/api/rest/2.0/ai/answer/create",
+                    json={
+                        "query": probe_query,
+                        "metadata_identifier": model_guid,
+                    },
+                    timeout=60,
+                )
+                if response.status_code == 200:
+                    data = response.json() or {}
+                    if data.get("session_identifier") and data.get("tokens"):
+                        _log(f"   [OK] Model answer-ready after {attempt} probe(s)")
+                        if session_logger:
+                            session_logger.log(
+                                "thoughtspot",
+                                "model answer-ready",
+                                model_guid=model_guid,
+                                probe_query=probe_query,
+                                attempts=attempt,
+                            )
+                        return True
+                    last_error = (
+                        "answer response missing session/tokens "
+                        f"(keys={list(data.keys())})"
+                    )
+                else:
+                    last_error = f"HTTP {response.status_code}: {response.text[:300]}"
+            except Exception as exc:
+                last_error = f"{type(exc).__name__}: {exc}"
+
+            _log(
+                f"   ⏳ Model answer-ready poll {attempt}: not ready "
+                f"({last_error[:180]})"
+            )
+            if session_logger:
+                session_logger.log(
+                    "thoughtspot",
+                    "model answer-ready poll",
+                    model_guid=model_guid,
+                    probe_query=probe_query,
+                    attempt=attempt,
+                    ready=False,
+                    error=last_error[:1000],
+                    timeout_seconds=timeout_seconds,
+                    poll_interval_seconds=poll_interval_seconds,
+                )
+
+            if time.time() >= deadline:
+                _log("   ⚠️  Model answer-ready check timed out; continuing to liveboard creation")
+                return False
+            time.sleep(min(poll_interval_seconds, max(0, deadline - time.time())))
+
     def parse_ddl(self, ddl: str) -> Tuple[Dict, List]:
         """
         Parse DDL to extract table definitions and foreign key relationships
@@ -1693,6 +1789,17 @@ class ThoughtSpotDeployer:
             print(f"   ⚠️  Could not check existing connections: {e}")
             return None
 
+    def _metadata_header(self, metadata_object: Dict) -> Dict:
+        """Return the metadata header regardless of ThoughtSpot API response shape."""
+        if not isinstance(metadata_object, dict):
+            return {}
+        return (
+            metadata_object.get("metadata_header")
+            or metadata_object.get("header")
+            or metadata_object.get("response", {}).get("header")
+            or {}
+        )
+
     def _parse_tml_edoc(self, edoc):
         """Parse ThoughtSpot TML export content whether the API returns YAML, JSON, or a dict."""
         if isinstance(edoc, dict):
@@ -1712,9 +1819,10 @@ class ThoughtSpotDeployer:
                                   connection_fqn: str = None) -> Dict:
         """Find an existing ThoughtSpot logical table by name and optional backing table context."""
         try:
-            response = self.session.get(
+            response = self.session.post(
                 f"{self.base_url}/api/rest/2.0/metadata/search",
-                params={
+                headers=self.headers,
+                json={
                     "metadata": [{"type": "LOGICAL_TABLE", "identifier": table_name}],
                     "record_size": 100,
                 },
@@ -1725,16 +1833,20 @@ class ThoughtSpotDeployer:
             candidates = response.json() or []
             exact_name = table_name.upper()
             for candidate in candidates:
+                header = self._metadata_header(candidate)
                 candidate_name = (
                     candidate.get("metadata_name")
                     or candidate.get("name")
-                    or candidate.get("header", {}).get("name")
+                    or header.get("name")
+                    or header.get("display_name")
+                    or header.get("displayName")
                     or ""
                 )
                 candidate_guid = (
                     candidate.get("metadata_id")
                     or candidate.get("id_guid")
-                    or candidate.get("header", {}).get("id_guid")
+                    or header.get("id_guid")
+                    or header.get("id")
                 )
                 if candidate_name.upper() != exact_name or not candidate_guid:
                     continue
@@ -2670,29 +2782,38 @@ class ThoughtSpotDeployer:
                             chunk_tmls,
                             create_new,
                         )
-                        if isinstance(result, requests.Response) and result.status_code in retriable_statuses:
-                            if create_new:
-                                resolved = _resolve_existing_tables_after_timeout(
-                                    chunk_names,
-                                    phase_label=f"{chunk_label} retry",
-                                )
-                            else:
-                                resolved = _verify_table_updates_after_timeout(
-                                    chunk_names,
-                                    chunk_tmls,
-                                    f"{chunk_label} retry",
-                                )
+                        if isinstance(result, requests.Response):
+                            if result.status_code in retriable_statuses:
+                                if create_new:
+                                    resolved = _resolve_existing_tables_after_timeout(
+                                        chunk_names,
+                                        phase_label=f"{chunk_label} retry",
+                                    )
+                                else:
+                                    resolved = _verify_table_updates_after_timeout(
+                                        chunk_names,
+                                        chunk_tmls,
+                                        f"{chunk_label} retry",
+                                    )
+                                all_objects.update(resolved)
+                                missing_names = [name for name in chunk_names if name not in resolved]
+                                if missing_names:
+                                    error = (
+                                        f"{chunk_label} timed out and {len(missing_names)} table(s) "
+                                        f"could not be verified after full-payload retry: {', '.join(missing_names)}"
+                                    )
+                                    _record_import_problem(error, fatal_errors=fatal_errors)
+                                    if fatal_errors:
+                                        return None
+                                continue
+                        elif result is None and create_new:
+                            resolved = _resolve_existing_tables_after_timeout(
+                                chunk_names,
+                                phase_label=f"{chunk_label} retry empty response",
+                            )
                             all_objects.update(resolved)
-                            missing_names = [name for name in chunk_names if name not in resolved]
-                            if missing_names:
-                                error = (
-                                    f"{chunk_label} timed out and {len(missing_names)} table(s) "
-                                    f"could not be verified after full-payload retry: {', '.join(missing_names)}"
-                                )
-                                _record_import_problem(error, fatal_errors=fatal_errors)
-                                if fatal_errors:
-                                    return None
-                            continue
+                            if all(name in resolved for name in chunk_names):
+                                continue
 
                     if isinstance(result, requests.Response):
                         if create_new:
@@ -3139,6 +3260,14 @@ class ThoughtSpotDeployer:
                             model_columns = liveboard_context['model_columns']
                             matrix_config = liveboard_context['matrix_config']
 
+                            log_progress("   Checking model answer readiness before liveboard creation...")
+                            self.wait_for_model_answer_ready(
+                                model_guid=model_guid,
+                                model_columns=model_columns,
+                                log_callback=log_progress,
+                                session_logger=_slog,
+                            )
+
                             log_progress("   Step 1/2: MCP creating liveboard...")
                             log_progress(f"   [MCP] Model: {model_name}, GUID: {model_guid}")
                             log_progress(f"   [MCP] Use case: {use_case or 'General Analytics'}")