032: fix final sync→async cascade (dataset_review, parsing, validation)

- _parsing.py: parse_superset_link + _recover_dataset_binding → async
    (await get_dashboard_detail, get_chart)
  - dataset_review/orchestrator.py: start_session + _build_recovery_bootstrap → async
    (await get_dataset_detail, parse_superset_link)
  - _routes.py (dataset_review): await orchestrator.start_session()
  - validation_tasks.py: await parse_superset_link + get_dashboard_detail

backend/src/api/routes/dataset_review_pkg/_routes.py

@@ -161,7 +161,7 @@ async def start_session(
             },
         )
         try:
-            result = orchestrator.start_session(
+            result = await orchestrator.start_session(
                 StartSessionCommand(
                     user=current_user,
                     environment_id=request.environment_id,

backend/src/api/routes/validation_tasks.py

@@ -412,7 +412,7 @@ async def parse_superset_url(
         from ...core.utils.superset_context_extractor import SupersetContextExtractor
 
         extractor = SupersetContextExtractor(environment=env)
-        parsed = extractor.parse_superset_link(payload.url)
+        parsed = await extractor.parse_superset_link(payload.url)
 
         dashboard_id = str(parsed.dashboard_id) if parsed.dashboard_id is not None else None
         title: str | None = None
@@ -420,7 +420,7 @@ async def parse_superset_url(
         # Resolve dashboard title from Superset API when possible
         if parsed.dashboard_id:
             try:
-                dashboard = extractor.client.get_dashboard_detail(parsed.dashboard_id)
+                dashboard = await extractor.client.get_dashboard_detail(parsed.dashboard_id)
                 title = dashboard.get("dashboard_title", str(parsed.dashboard_id))
             except Exception:
                 title = str(parsed.dashboard_id)

backend/src/core/utils/superset_context_extractor/_parsing.py

@@ -25,7 +25,7 @@ class SupersetContextParsingMixin:
     # @POST  returns resolved dataset/dashboard context, preserving explicit partial-recovery state if some identifiers cannot be confirmed.
     # @SIDE_EFFECT  may issue Superset API reads to resolve dataset references from dashboard or chart URLs.
     # @DATA_CONTRACT  Input[link:str] -> Output[SupersetParsedContext]
-    def parse_superset_link(self, link: str) -> SupersetParsedContext:
+    async def parse_superset_link(self, link: str) -> SupersetParsedContext:
         with belief_scope("SupersetContextExtractor.parse_superset_link"):
             normalized_link = str(link or "").strip()
             if not normalized_link:
@@ -106,7 +106,7 @@ class SupersetContextParsingMixin:
                             != "dashboard_permalink_dataset_binding_unresolved"
                         ]
                         dataset_id, unresolved_references = (
-                            self._recover_dataset_binding_from_dashboard(
+                            await self._recover_dataset_binding_from_dashboard(
                                 dashboard_id=dashboard_id,
                                 dataset_ref=dataset_ref,
                                 unresolved_references=unresolved_references,
@@ -123,7 +123,7 @@ class SupersetContextParsingMixin:
                             != "dashboard_permalink_dataset_binding_unresolved"
                         ]
                         try:
-                            chart_payload = self.client.get_chart(chart_id)
+                            chart_payload = await self.client.get_chart(chart_id)
                             chart_data = (
                                 chart_payload.get("result", chart_payload)
                                 if isinstance(chart_payload, dict)
@@ -170,7 +170,7 @@ class SupersetContextParsingMixin:
                 )
                 # Resolve dashboard detail first — handles both numeric ID and slug,
                 # ensuring dashboard_id is available for the native_filters_key fetch below.
-                dashboard_detail = self.client.get_dashboard_detail(
+                dashboard_detail = await self.client.get_dashboard_detail(
                     resolved_dashboard_ref
                 )
                 resolved_dashboard_id = dashboard_detail.get("id")
@@ -317,13 +317,13 @@ class SupersetContextParsingMixin:
     # #region SupersetContextParsingMixin._recover_dataset_binding_from_dashboard [TYPE Function] [C:3]
     # @PURPOSE: Recover a dataset binding from resolved dashboard context while preserving explicit unresolved markers.
     # @RELATION CALLS -> [EXT:method:SupersetClient.get_dashboard_detail]
-    def _recover_dataset_binding_from_dashboard(
+    async def _recover_dataset_binding_from_dashboard(
         self,
         dashboard_id: int,
         dataset_ref: str | None,
         unresolved_references: list[str],
     ) -> tuple[int | None, list[str]]:
-        dashboard_detail = self.client.get_dashboard_detail(dashboard_id)
+        dashboard_detail = await self.client.get_dashboard_detail(dashboard_id)
         datasets = dashboard_detail.get("datasets") or []
         if datasets:
             first_dataset = datasets[0]

backend/src/services/dataset_review/orchestrator.py

@@ -123,7 +123,7 @@ class DatasetReviewOrchestrator:
     # @SIDE_EFFECT  persists session and may enqueue recovery task.
     # @DATA_CONTRACT  Input[StartSessionCommand] -> Output[StartSessionResult]
     # @INVARIANT  no cross-user session leakage occurs; session and follow-up task remain owned by the authenticated user.
-    def start_session(self, command: StartSessionCommand) -> StartSessionResult:
+    async def start_session(self, command: StartSessionCommand) -> StartSessionResult:
         with belief_scope("DatasetReviewOrchestrator.start_session"):
             normalized_source_kind = str(command.source_kind or "").strip()
             normalized_source_input = str(command.source_input or "").strip()
@@ -155,7 +155,7 @@ class DatasetReviewOrchestrator:
 
             if normalized_source_kind == "superset_link":
                 extractor = SupersetContextExtractor(environment)
-                parsed_context = extractor.parse_superset_link(normalized_source_input)
+                parsed_context = await extractor.parse_superset_link(normalized_source_input)
                 dataset_ref = parsed_context.dataset_ref
                 dataset_id = parsed_context.dataset_id
                 dashboard_id = parsed_context.dashboard_id
@@ -190,8 +190,8 @@ class DatasetReviewOrchestrator:
             template_variables: list[TemplateVariable] = []
             execution_mappings: list[ExecutionMapping] = []
             if normalized_source_kind == "superset_link" and parsed_context is not None:
-                recovered_filters, template_variables, execution_mappings, findings = (
+                    recovered_filters, template_variables, execution_mappings, findings = (
-                    self._build_recovery_bootstrap(
+                        await self._build_recovery_bootstrap(
                         environment=environment,
                         session=persisted_session,
                         parsed_context=parsed_context,
@@ -451,7 +451,7 @@ class DatasetReviewOrchestrator:
     # @PRE  session belongs to the just-created review aggregate and parsed_context was produced for the same environment scope.
     # @POST  Returns bootstrap imported filters, template variables, execution mappings, and updated findings without persisting them directly.
     # @SIDE_EFFECT  Performs Superset reads through the extractor and may append warning findings for incomplete recovery.
-    def _build_recovery_bootstrap(
+    async def _build_recovery_bootstrap(
         self,
         environment,
         session: DatasetReviewSession,
@@ -487,7 +487,7 @@ class DatasetReviewOrchestrator:
             try:
                 dataset_payload = parsed_context.dataset_payload
                 if not isinstance(dataset_payload, dict):
-                    dataset_payload = extractor.client.get_dataset_detail(session_record.dataset_id)
+                    dataset_payload = await extractor.client.get_dataset_detail(session_record.dataset_id)
                 discovered_variables = extractor.discover_template_variables(dataset_payload)
                 template_variables = [
                     TemplateVariable(