032: Phase 4 US2 — Translate plugin fully async

T022-T028: All translate methods async.
- _llm_async_http.py created (httpx.AsyncClient+asyncio.sleep)
- Old _llm_http.py and preview_llm_client.py preserved (tombstone later)
- superset_executor, preview, executor, run_source, llm_call all async

RATIONALE: httpx.AsyncClient + asyncio.sleep instead of time.sleep.
REJECTED: AsyncOpenAI SDK — doesn't support custom base_url.

backend/src/plugins/translate/_batch_proc.py

@@ -46,7 +46,7 @@ class BatchProcessingService:
     # @PRE job and batch_rows are valid.
     # @POST TranslationBatch and TranslationRecord rows are created.
     # @SIDE_EFFECT LLM API call; DB writes.
-    def process_batch(
+    async def process_batch(
         self, job: TranslationJob, run_id: str, batch_index: int,
         batch_rows: list[dict[str, Any]],
         dict_snapshot_hash: str | None = None, config_hash: str | None = None,
@@ -76,7 +76,7 @@ class BatchProcessingService:
             # Count cache hits: pre_rows that have _cached_lang_values (served from cache, not same-lang/approved)
             result["cache_hits"] = sum(1 for r in pre_rows if r.get("_cached_lang_values"))
             if llm_rows:
-                llm_res = self._process_llm(job, run_id, llm_rows, dict_matches, bid, tls)
+                llm_res = await self._process_llm(job, run_id, llm_rows, dict_matches, bid, tls)
                 for k in ("successful", "failed", "skipped", "retries"):
                     result[k] += llm_res.get(k, 0)
 
@@ -205,7 +205,7 @@ class BatchProcessingService:
             count += 1
         return count
 
-    def _process_llm(self, job, run_id, rows_for_llm, dict_matches, bid, tls):
+    async def _process_llm(self, job, run_id, rows_for_llm, dict_matches, bid, tls):
         # Resolve provider token config (DB values take priority over PROVIDER_DEFAULTS)
         token_config = {"model": None, "context_window": None, "max_output_tokens": None}
         if job.provider_id:
@@ -239,7 +239,7 @@ class BatchProcessingService:
             },
         )
 
-        result = self._llm_service.call_llm_for_batch(
+        result = await self._llm_service.call_llm_for_batch(
             job=job, run_id=run_id, batch_rows=rows_for_llm,
             dict_matches=dict_matches, batch_id=bid,
             max_tokens=tb["max_output_needed"],

backend/src/plugins/translate/_llm_async_http.py

@@ -0,0 +1,243 @@
+# #region LLMAsyncHttpClient [C:4] [TYPE Module] [SEMANTICS translate, llm, http, openai, async, retry, rate-limit]
+# @BRIEF Async HTTP client for OpenAI-compatible LLM API calls using httpx.AsyncClient
+#         with rate-limit handling and structured output fallback.
+# @LAYER Infrastructure
+# @RELATION DEPENDS_ON -> [EXT:httpx:AsyncClient]
+# @PRE Valid API endpoint, key, model, and prompt.
+# @POST Returns (response text, finish_reason) tuple.
+# @SIDE_EFFECT Async HTTP POST to LLM API with optional retry on 429.
+# @RATIONALE Async migration of _llm_http.py to use httpx.AsyncClient instead of sync requests.
+#            Uses asyncio.sleep for 429 backoff instead of time.sleep.
+# @REJECTED Keeping sync requests.post — would block async event loop during LLM calls.
+
+import asyncio
+import os
+from typing import Any
+
+import httpx
+
+from ...core.logger import logger
+
+# Module-level httpx client, lazily initialized for connection reuse
+_http_client: httpx.AsyncClient | None = None
+
+# Default provider and max_tokens constants
+DEFAULT_PROVIDER_TYPE: str = "openai"
+DEFAULT_MAX_TOKENS: int = 8192
+
+
+# #region _get_verify [C:1] [TYPE Function] [SEMANTICS translate, ssl, verify]
+# @BRIEF Resolve SSL verification path from LLM_SSL_VERIFY env var.
+# @RATIONALE Используем capath=/etc/ssl/certs/ вместо cafile, потому что
+#   OpenSSL 3.x не использует intermediate CA сертификаты из cafile для
+#   построения цепочки (verify code 20). capath с хеш-симлинками работает
+#   корректно (verify code 0).
+# @REJECTED cafile отвергнут — OpenSSL 3.x не использует intermediate CA
+#   из единого bundle-файла. Только capath с хеш-симлинками даёт code 0.
+# @POST Returns path to /etc/ssl/certs/ when enabled, False when disabled.
+def _get_verify() -> str | bool:
+    raw = os.getenv("LLM_SSL_VERIFY", "true").strip().lower()
+    if raw in ("false", "0", "no", "off"):
+        return False
+    return "/etc/ssl/certs/"
+# #endregion _get_verify
+
+
+# #region _get_http_client [C:1] [TYPE Function]
+# @BRIEF Get or create the module-level httpx.AsyncClient singleton.
+# @POST Returns httpx.AsyncClient with SSL verify and 180s timeout.
+async def _get_http_client() -> httpx.AsyncClient:
+    global _http_client
+    if _http_client is None:
+        _http_client = httpx.AsyncClient(
+            verify=_get_verify(),
+            timeout=httpx.Timeout(180.0),
+        )
+    return _http_client
+# #endregion _get_http_client
+
+
+# #region call_openai_compatible [C:3] [TYPE Function] [SEMANTICS translate, llm, http, openai, async]
+# @BRIEF Call OpenAI-compatible API asynchronously with rate-limit handling and structured output fallback.
+# @PRE Valid API endpoint, key, model, and prompt.
+# @POST Returns (response text, finish_reason).
+# @SIDE_EFFECT Async HTTP POST to LLM API.
+async def call_openai_compatible(
+    base_url: str,
+    api_key: str,
+    model: str,
+    prompt: str,
+    provider_type: str = "openai",
+    max_tokens: int = 8192,
+    disable_reasoning: bool = False,
+) -> tuple[str, str | None]:
+    """Call OpenAI-compatible API for batch translation (async)."""
+    if not base_url:
+        raise ValueError("LLM provider has no base_url configured")
+
+    url = f"{base_url.rstrip('/')}/chat/completions"
+    headers = {
+        "Authorization": f"Bearer {api_key}",
+        "Content-Type": "application/json",
+    }
+    system_content = (
+        "You are a database content translation assistant. "
+        "Translate the provided text accurately, preserving data semantics. "
+        "Respond directly with ONLY the JSON result. "
+        "Do NOT include any reasoning, thinking, chain-of-thought, analysis, "
+        "or explanation. Output ONLY valid JSON."
+    )
+
+    payload: dict[str, Any] = {
+        "model": model,
+        "messages": [
+            {"role": "system", "content": system_content},
+            {"role": "user", "content": prompt},
+        ],
+        "temperature": 0.1,
+        "max_tokens": max_tokens,
+    }
+
+    if provider_type in ("openai", "openai_compatible", "kilo", "openrouter", "litellm"):
+        if not disable_reasoning:
+            payload["response_format"] = {"type": "json_object"}
+
+    if disable_reasoning:
+        if provider_type not in ("kilo", "openrouter", "litellm"):
+            payload["reasoning_effort"] = "none"
+        payload["max_tokens"] = max_tokens
+
+    logger.reason(
+        f"LLM request url={base_url} model={payload.get('model')} "
+        f"provider_type={provider_type} "
+        f"response_format={'yes' if 'response_format' in payload else 'no'} "
+        f"prompt_len={len(prompt)}"
+    )
+
+    response, response_text = await _do_http_request(url, headers, payload)
+    await _handle_response_format_fallback(response, response_text, payload, url, headers)
+
+    if not response.ok:
+        logger.explore(
+            f"LLM API error status={response.status_code} "
+            f"model={payload.get('model')} "
+            f"body={response_text[:2000]}"
+        )
+    response.raise_for_status()
+    data = response.json()
+
+    choices = data.get("choices", [])
+    if not choices:
+        logger.explore("LLM returned no choices", extra={
+            "src": "executor", "response_keys": list(data.keys()),
+            "response_preview": str(data)[:2000],
+        })
+        raise ValueError("LLM returned no choices")
+
+    try:
+        finish_reason = choices[0].get("finish_reason") or "none"
+        msg = choices[0].get("message") or {}
+    except (TypeError, AttributeError) as e:
+        logger.explore("TypeError processing LLM response choices", extra={
+            "src": "executor_diag", "error": str(e),
+            "choices_0_type": type(choices[0]).__name__ if choices else "N/A",
+            "choices_0_repr": repr(choices[0])[:2000] if choices else "N/A",
+            "data_type": type(data).__name__, "data_preview": str(data)[:2000],
+        })
+        raise ValueError(f"LLM response processing failed: {e}")
+
+    # Log provider token usage for batch sizing calibration
+    usage = data.get("usage") or {}
+    if usage:
+        logger.reason(
+            "LLM provider usage",
+            {
+                "prompt_tokens": usage.get("prompt_tokens"),
+                "completion_tokens": usage.get("completion_tokens"),
+                "total_tokens": usage.get("total_tokens"),
+                "finish_reason": finish_reason,
+                "max_tokens_sent": max_tokens,
+                "chars_sent": len(prompt),
+            },
+        )
+
+    refusal = msg.get("refusal") if isinstance(msg, dict) else None
+    if refusal:
+        logger.explore("LLM refused to respond", extra={
+            "src": "executor", "refusal": str(refusal)[:500], "finish_reason": finish_reason,
+        })
+        raise ValueError(f"LLM refused to respond: {refusal}")
+
+    content = msg.get("content") if isinstance(msg, dict) else ""
+    if not content and isinstance(msg, dict):
+        content = msg.get("content") or ""
+
+    logger.reason(
+        f"LLM response finish_reason={finish_reason} content_len={len(content)} "
+        f"msg_keys={list(msg.keys()) if isinstance(msg, dict) else []}"
+    )
+    if not content:
+        logger.explore("LLM returned empty content", extra={
+            "src": "executor", "finish_reason": finish_reason,
+            "msg_keys": list(msg.keys()) if isinstance(msg, dict) else [],
+            "response_preview": str(data)[:2000],
+        })
+        raise ValueError("LLM returned empty content")
+
+    return content, finish_reason
+# #endregion call_openai_compatible
+
+
+# #region _do_http_request [C:1] [TYPE Function]
+async def _do_http_request(url: str, headers: dict, payload: dict) -> tuple[httpx.Response, str]:
+    """Make async HTTP POST with rate-limit (429) retry handling."""
+    client = await _get_http_client()
+    _max_retry_429 = 3
+    _retry_count_429 = 0
+    while _retry_count_429 < _max_retry_429:
+        response = await client.post(url, headers=headers, json=payload)
+        response_text = response.text
+        if response.status_code == 429:
+            _retry_count_429 += 1
+            retry_after = response.headers.get("Retry-After")
+            if retry_after:
+                try:
+                    wait = int(retry_after)
+                except (ValueError, TypeError):
+                    wait = 2 ** _retry_count_429
+            else:
+                wait = 2 ** _retry_count_429
+            logger.explore(f"Rate limited (429), retry {_retry_count_429}/{_max_retry_429} after {wait}s",
+                           extra={"src": "executor", "retry_after": retry_after, "wait": wait})
+            await asyncio.sleep(wait)
+            if _retry_count_429 >= _max_retry_429:
+                break
+        else:
+            break
+    return response, response_text
+# #endregion _do_http_request
+
+
+# #region _handle_response_format_fallback [C:1] [TYPE Function]
+async def _handle_response_format_fallback(
+    response: httpx.Response, response_text: str, payload: dict, url: str, headers: dict,
+) -> None:
+    """Handle 400 errors from structured_outputs not being supported."""
+    _patterns = ("response_format", "structured_outputs", "structured", "json_object")
+    if (
+        not response.ok
+        and response.status_code == 400
+        and any(p in (response_text or "").lower() for p in _patterns)
+    ):
+        client = await _get_http_client()
+        logger.explore("Structured outputs not supported, retrying without response_format",
+                       extra={"src": "executor"})
+        payload.pop("response_format", None)
+        new_response = await client.post(url, headers=headers, json=payload)
+        # Mutate the original response object with new data
+        response.status_code = new_response.status_code
+        response._content = new_response.content
+        response.encoding = new_response.encoding
+        response.headers = new_response.headers
+# #endregion _handle_response_format_fallback
+# #endregion LLMAsyncHttpClient

backend/src/plugins/translate/_llm_call.py

@@ -16,9 +16,9 @@
 #            _llm_parse.py to meet INV_7 module limit (< 400 lines).
 # @REJECTED Single monolithic call_llm_for_batch at 327 lines — split into focused sub-methods.
 
+import asyncio
 from datetime import UTC, datetime
 import json
-import time
 from typing import Any
 import uuid
 
@@ -28,7 +28,7 @@ from ...core.logger import belief_scope, logger
 from ...models.translate import TranslationBatch, TranslationJob, TranslationLanguage, TranslationRecord
 from ...services.llm_prompt_templates import render_prompt
 from ...services.llm_provider import LLMProviderService
-from ._llm_http import call_openai_compatible
+from ._llm_async_http import call_openai_compatible
 from ._llm_parse import parse_llm_response
 from ._utils import _enforce_dictionary
 from .preview import DEFAULT_EXECUTION_PROMPT_TEMPLATE
@@ -49,7 +49,7 @@ class LLMTranslationService:
     # @PRE job has valid provider_id. batch_rows is non-empty.
     # @POST Returns dict with successful/failed/skipped counts.
     # @SIDE_EFFECT HTTP call to LLM provider; DB writes.
-    def call_llm_for_batch(
+    async def call_llm_for_batch(
         self, job: TranslationJob, run_id: str,
         batch_rows: list[dict[str, Any]], dict_matches: list[dict[str, Any]],
         batch_id: str, max_tokens: int = 8192, _recursion_depth: int = 0,
@@ -68,7 +68,7 @@ class LLMTranslationService:
             target_languages = self._resolve_target_languages(job)
             prompt = self._build_prompt(job, batch_rows, dictionary_section, target_languages)
 
-            llm_response, finish_reason, retries, last_error = self._call_llm_with_retry(
+            llm_response, finish_reason, retries, last_error = await self._call_llm_with_retry(
                 job, prompt, batch_id, max_tokens,
             )
             if llm_response is None:
@@ -96,7 +96,7 @@ class LLMTranslationService:
                             f"Retrying only {len(remaining)}/{len(batch_rows)} missing rows",
                             {"batch_id": batch_id, "recovered": len(recovered_ids), "remaining": len(remaining)},
                         )
-                        return self._retry_missing_rows(
+                        return await self._retry_missing_rows(
                             job, run_id, remaining, dict_matches,
                             batch_id, max_tokens, _recursion_depth,
                         )
@@ -116,8 +116,8 @@ class LLMTranslationService:
                             "depth": _recursion_depth,
                         },
                     )
-                    return self._split_and_retry(job, run_id, batch_rows, dict_matches,
-                                                  batch_id, max_tokens, _recursion_depth, retries)
+                    return await self._split_and_retry(job, run_id, batch_rows, dict_matches,
+                                                       batch_id, max_tokens, _recursion_depth, retries)
                 logger.explore("Truncation recursion depth exceeded", {"batch_id": batch_id, "depth": _recursion_depth})
 
             try:
@@ -187,7 +187,7 @@ class LLMTranslationService:
     # #region _call_llm_with_retry [C:3] [TYPE Function] [SEMANTICS translate, llm, retry]
     # @BRIEF Call LLM with retry loop (max 3 attempts, exponential backoff).
     # @SIDE_EFFECT HTTP calls to LLM provider on each attempt.
-    def _call_llm_with_retry(self, job, prompt, batch_id, max_tokens):
+    async def _call_llm_with_retry(self, job, prompt, batch_id, max_tokens):
         llm_response = None
         last_error = None
         retries = 0
@@ -195,7 +195,7 @@ class LLMTranslationService:
         logger.reason(f"LLM retry loop start", {"batch_id": batch_id, "max_retries": MAX_RETRIES_PER_BATCH, "prompt_len": len(prompt)})
         for attempt in range(1, MAX_RETRIES_PER_BATCH + 1):
             try:
-                llm_response, finish_reason = self.call_llm(job, prompt, max_tokens=max_tokens)
+                llm_response, finish_reason = await self.call_llm(job, prompt, max_tokens=max_tokens)
                 logger.reason(
                     f"LLM call succeeded (attempt {attempt})", {
                         "batch_id": batch_id, "finish_reason": finish_reason,
@@ -208,7 +208,7 @@ class LLMTranslationService:
                 retries += 1
                 logger.explore(f"LLM call failed (attempt {attempt})", {"batch_id": batch_id, "error": last_error})
                 if attempt < MAX_RETRIES_PER_BATCH:
-                    time.sleep(2 ** attempt)
+                    await asyncio.sleep(2 ** attempt)
         if llm_response is None:
             logger.explore(f"All LLM retries exhausted", {"batch_id": batch_id, "retries": retries, "last_error": last_error})
         return llm_response, finish_reason, retries, last_error
@@ -233,7 +233,7 @@ class LLMTranslationService:
     # #region _split_and_retry [C:3] [TYPE Function] [SEMANTICS translate, llm, split, retry]
     # @BRIEF Binary-split a batch and retry each half recursively.
     # @SIDE_EFFECT Creates two child TranslationBatch rows; recursive LLM calls.
-    def _split_and_retry(self, job, run_id, batch_rows, dict_matches, batch_id, max_tokens, depth, retries):
+    async def _split_and_retry(self, job, run_id, batch_rows, dict_matches, batch_id, max_tokens, depth, retries):
         mid = len(batch_rows) // 2
         logger.explore("LLM output truncated — splitting batch",
                        {"batch_id": batch_id, "batch_size": len(batch_rows), "split_at": mid, "depth": depth})
@@ -257,10 +257,10 @@ class LLMTranslationService:
         self.db.add_all([left_batch, right_batch])
         self.db.flush()
 
-        left = self.call_llm_for_batch(job, run_id, batch_rows[:mid], dict_matches,
-                                        left_batch.id, max_tokens, depth + 1)
-        right = self.call_llm_for_batch(job, run_id, batch_rows[mid:], dict_matches,
-                                         right_batch.id, max_tokens, depth + 1)
+        left = await self.call_llm_for_batch(job, run_id, batch_rows[:mid], dict_matches,
+                                               left_batch.id, max_tokens, depth + 1)
+        right = await self.call_llm_for_batch(job, run_id, batch_rows[mid:], dict_matches,
+                                                right_batch.id, max_tokens, depth + 1)
 
         # Finalise child batch stats
         left_batch.completed_at = datetime.now(UTC)
@@ -355,7 +355,7 @@ class LLMTranslationService:
     # @PRE missing_rows is a subset of the original batch rows, or empty.
     # @POST Returns dict with successful/failed/skipped counts from the sub-batch.
     # @SIDE_EFFECT Creates TranslationBatch for the retry sub-batch; may recurse.
-    def _retry_missing_rows(self, job, run_id, missing_rows, dict_matches, _batch_id, max_tokens, depth):
+    async def _retry_missing_rows(self, job, run_id, missing_rows, dict_matches, _batch_id, max_tokens, depth):
         if not missing_rows:
             return {"successful": 0, "failed": 0, "skipped": 0, "retries": 0}
 
@@ -367,7 +367,7 @@ class LLMTranslationService:
         self.db.add(sub_batch)
         self.db.flush()
 
-        result = self.call_llm_for_batch(
+        result = await self.call_llm_for_batch(
             job, run_id, missing_rows, dict_matches,
             sub_batch.id, max_tokens, depth,
         )
@@ -483,7 +483,7 @@ class LLMTranslationService:
 
     # #region call_llm [C:3] [TYPE Function] [SEMANTICS translate, llm, call]
     # @BRIEF Route to provider-specific LLM call implementation.
-    def call_llm(self, job: TranslationJob, prompt: str, max_tokens: int = 8192) -> tuple[str, str | None]:
+    async def call_llm(self, job: TranslationJob, prompt: str, max_tokens: int = 8192) -> tuple[str, str | None]:
         with belief_scope("LLMTranslationService.call_llm"):
             if not job.provider_id:
                 raise ValueError("Job has no LLM provider configured")
@@ -508,7 +508,7 @@ class LLMTranslationService:
 
             if provider_type not in ("openai", "openai_compatible", "openrouter", "kilo", "litellm"):
                 raise ValueError(f"Unsupported provider type '{provider_type}'")
-            result = call_openai_compatible(
+            result = await call_openai_compatible(
                 base_url=provider.base_url, api_key=api_key, model=model, prompt=prompt,
                 provider_type=provider_type, max_tokens=max_tokens, disable_reasoning=disable_reasoning,
             )
@@ -522,10 +522,10 @@ class LLMTranslationService:
     # #endregion call_llm
 
     # #region call_openai_compatible [C:1] [TYPE Function] [SEMANTICS translate, llm, compat]
-    # @BRIEF Backward-compat delegation to _llm_http.call_openai_compatible.
+    # @BRIEF Backward-compat delegation to _llm_async_http.call_openai_compatible.
     @staticmethod
-    def call_openai_compatible(*a, **kw):
-        return call_openai_compatible(*a, **kw)
+    async def call_openai_compatible(*a, **kw):
+        return await call_openai_compatible(*a, **kw)
     # #endregion call_openai_compatible
 
     # #region _parse_llm_response [C:1] [TYPE Function] [SEMANTICS translate, llm, compat]

backend/src/plugins/translate/_run_service.py

@@ -39,8 +39,8 @@ class RunExecutionService:
         self._preview_edits_cache: dict[str, dict[str, str]] | None = None
 
     # -- Source fetching (thin wrappers) --
-    def _fetch_source_rows(self, job_id: str, run_id: str) -> list[dict[str, Any]]:
-        return fetch_source_rows(self.db, self.config_manager, job_id, run_id)
+    async def _fetch_source_rows(self, job_id: str, run_id: str) -> list[dict[str, Any]]:
+        return await fetch_source_rows(self.db, self.config_manager, job_id, run_id)
 
     @staticmethod
     def _extract_chart_data_rows(response):

backend/src/plugins/translate/_run_source.py

@@ -25,7 +25,7 @@ MAX_ROWS_PER_RUN = 10000
 
 # #region fetch_source_rows [C:3] [TYPE Function] [SEMANTICS translate, source, fetch]
 # @BRIEF Fetch source rows from Superset datasource or preview session fallback.
-def fetch_source_rows(db: Session, config_manager: ConfigManager, job_id: str, run_id: str) -> list[dict[str, Any]]:
+async def fetch_source_rows(db: Session, config_manager: ConfigManager, job_id: str, run_id: str) -> list[dict[str, Any]]:
     """Fetch source rows from Superset datasource or preview session fallback."""
     with belief_scope("RunSourceFetcher.fetch_source_rows"):
         job = db.query(TranslationJob).filter(TranslationJob.id == job_id).first()
@@ -45,7 +45,7 @@ def fetch_source_rows(db: Session, config_manager: ConfigManager, job_id: str, r
                 if env_config:
                     from ...core.superset_client import SupersetClient
                     client = SupersetClient(env_config)
-                    dataset_detail = client.get_dataset_detail(int(job.source_datasource_id))
+                    dataset_detail = await client.get_dataset_detail(int(job.source_datasource_id))
                     query_context = client.build_dataset_preview_query_context(
                         dataset_id=int(job.source_datasource_id), dataset_record=dataset_detail,
                         template_params={}, effective_filters=[],
@@ -59,7 +59,7 @@ def fetch_source_rows(db: Session, config_manager: ConfigManager, job_id: str, r
                     form_data = query_context.get("form_data", {})
                     form_data.pop("query_mode", None)
 
-                    response = client.network.request(
+                    response = await client.network.request(
                         method="POST", endpoint="/api/v1/chart/data",
                         data=json.dumps(query_context), headers={"Content-Type": "application/json"},
                     )

backend/src/plugins/translate/executor.py

@@ -59,7 +59,7 @@ class TranslationExecutor:
     # @PRE run is in PENDING or RUNNING status with valid job config.
     # @POST Run is populated with batches and records.
     # @SIDE_EFFECT LLM API calls; DB batch writes.
-    def execute_run(
+    async def execute_run(
         self, run: TranslationRun,
         llm_progress_callback: Callable[[str, int, int, int], None] | None = None,
         language_stats_map: dict[str, TranslationRunLanguageStats] | None = None,
@@ -70,7 +70,7 @@ class TranslationExecutor:
             raise ValueError(f"Job '{run.job_id}' not found for run '{run.id}'")
         logger.reason("Starting translation execution", {"run_id": run.id, "job_id": job.id})
 
-        run, source_rows, job = self._prepare_run(run, job)
+        run, source_rows, job = await self._prepare_run(run, job)
         if run.status != "RUNNING":
             return run
 
@@ -78,11 +78,11 @@ class TranslationExecutor:
         run.total_records = len(source_rows)
         logger.reason(f"Processing {len(batches)} batches", {"run_id": run.id, "total_rows": run.total_records})
 
-        return self._process_batches(run, job, batches, target_languages, language_stats_map)
+        return await self._process_batches(run, job, batches, target_languages, language_stats_map)
     # endregion execute_run
 
     # region _prepare_run [C:2] [TYPE Function]
-    def _prepare_run(self, run: TranslationRun, job: TranslationJob) -> tuple:
+    async def _prepare_run(self, run: TranslationRun, job: TranslationJob) -> tuple:
         """Prepare run: load preview edits, fetch source rows, filter new keys."""
         self._load_preview_edits(job.id)
         run.status = "RUNNING"
@@ -93,7 +93,7 @@ class TranslationExecutor:
         if run.config_snapshot and isinstance(run.config_snapshot, dict):
             full_translation = run.config_snapshot.get("full_translation", False)
 
-        source_rows = self._fetch_source_rows(job.id, run.id)
+        source_rows = await self._fetch_source_rows(job.id, run.id)
         if not source_rows:
             logger.explore("No source rows to translate", {"run_id": run.id})
             run.status = "COMPLETED"
@@ -125,13 +125,13 @@ class TranslationExecutor:
     # endregion _prepare_batches
 
     # region _process_batches [C:3] [TYPE Function]
-    def _process_batches(self, run: TranslationRun, job: TranslationJob, batches: list,
-                         target_languages: list[str],
-                         language_stats_map: dict[str, TranslationRunLanguageStats] | None = None) -> TranslationRun:
+    async def _process_batches(self, run: TranslationRun, job: TranslationJob, batches: list,
+                               target_languages: list[str],
+                               language_stats_map: dict[str, TranslationRunLanguageStats] | None = None) -> TranslationRun:
         """Process all batches: execute, insert to target, check cancellation."""
         successful_records = failed_records = skipped_records = cache_hits = 0
         for batch_idx, batch_rows in enumerate(batches):
-            batch_result = self._process_batch(
+            batch_result = await self._process_batch(
                 job=job, run_id=run.id, batch_index=batch_idx, batch_rows=batch_rows,
                 dict_snapshot_hash=run.dict_snapshot_hash, config_hash=run.config_hash,
             )
@@ -187,8 +187,8 @@ class TranslationExecutor:
         return run
 
     # -- Delegation methods (thin wrappers for test-patch compatibility) --
-    def _fetch_source_rows(self, job_id: str, run_id: str) -> list[dict[str, Any]]:
-        return self._run_service._fetch_source_rows(job_id, run_id)
+    async def _fetch_source_rows(self, job_id: str, run_id: str) -> list[dict[str, Any]]:
+        return await self._run_service._fetch_source_rows(job_id, run_id)
 
     def _filter_new_keys(self, job, run_id: str, source_rows: list) -> list:
         return self._run_service._filter_new_keys(job, run_id, source_rows)
@@ -241,9 +241,9 @@ class TranslationExecutor:
         return AdaptiveBatchSizer(self.db, self.config_manager).auto_size_batches(
             job, source_rows, target_languages, provider_info)
 
-    def _process_batch(self, job, run_id, batch_index, batch_rows, dict_snapshot_hash=None, config_hash=None) -> dict:
+    async def _process_batch(self, job, run_id, batch_index, batch_rows, dict_snapshot_hash=None, config_hash=None) -> dict:
         from ._batch_proc import BatchProcessingService
-        return BatchProcessingService(self.db, self.config_manager).process_batch(
+        return await BatchProcessingService(self.db, self.config_manager).process_batch(
             job, run_id, batch_index, batch_rows, dict_snapshot_hash, config_hash,
             preview_edits_cache=self._preview_edits_cache)
 
@@ -251,19 +251,19 @@ class TranslationExecutor:
         from ._batch_proc import BatchProcessingService
         BatchProcessingService(self.db, self.config_manager).insert_batch_to_target(job, batch_id, run_id)
 
-    def _call_llm_for_batch(self, job, run_id, batch_rows, dict_matches, batch_id, max_tokens=8192, _recursion_depth=0) -> dict:
+    async def _call_llm_for_batch(self, job, run_id, batch_rows, dict_matches, batch_id, max_tokens=8192, _recursion_depth=0) -> dict:
         from ._llm_call import LLMTranslationService
-        return LLMTranslationService(self.db).call_llm_for_batch(
+        return await LLMTranslationService(self.db).call_llm_for_batch(
             job, run_id, batch_rows, dict_matches, batch_id, max_tokens, _recursion_depth)
 
-    def _call_llm(self, job, prompt, max_tokens=8192) -> tuple:
+    async def _call_llm(self, job, prompt, max_tokens=8192) -> tuple:
         from ._llm_call import LLMTranslationService
-        return LLMTranslationService(self.db).call_llm(job, prompt, max_tokens)
+        return await LLMTranslationService(self.db).call_llm(job, prompt, max_tokens)
 
     @staticmethod
-    def _call_openai_compatible(base_url, api_key, model, prompt, provider_type="openai", max_tokens=8192, disable_reasoning=False) -> tuple:
+    async def _call_openai_compatible(base_url, api_key, model, prompt, provider_type="openai", max_tokens=8192, disable_reasoning=False) -> tuple:
         from ._llm_call import LLMTranslationService
-        return LLMTranslationService.call_openai_compatible(base_url, api_key, model, prompt, provider_type, max_tokens, disable_reasoning)
+        return await LLMTranslationService.call_openai_compatible(base_url, api_key, model, prompt, provider_type, max_tokens, disable_reasoning)
 
     @staticmethod
     def _parse_llm_response(response_text, expected_count, target_languages=None, finish_reason=None) -> dict:

backend/src/plugins/translate/preview.py

@@ -65,7 +65,7 @@ class TranslationPreview:
     # region preview_rows [TYPE Function]
     # @PURPOSE: Fetch sample rows, send to LLM, create preview session with per-language records.
     # @SIDE_EFFECT  Fetches data from Superset; calls LLM; creates DB rows.
-    def preview_rows(
+    async def preview_rows(
         self,
         job_id: str,
         sample_size: int = 10,
@@ -91,7 +91,7 @@ class TranslationPreview:
 
             config_hash = self._executor.compute_config_hash(job)
             dict_snapshot_hash = self._executor.compute_dict_snapshot_hash(job_id)
-            source_rows = self._executor.fetch_sample_rows(job=job, sample_size=sample_size, env_id=env_id)
+            source_rows = await self._executor.fetch_sample_rows(job=job, sample_size=sample_size, env_id=env_id)
             if not source_rows:
                 raise ValueError("No rows returned from datasource for preview")
 
@@ -116,7 +116,7 @@ class TranslationPreview:
                 meta["_detected_lang"] = detect_language(meta["source_text"], detector)
 
             t_llm = time.monotonic()
-            llm_response = self._executor.call_llm(
+            llm_response = await self._executor.call_llm(
                 job=job, prompt=prompt_data["prompt"], max_tokens=token_budget["max_output_needed"],
             )
             logger.reason(f"TIMING: LLM call: {time.monotonic() - t_llm:.2f}s", {})

backend/src/plugins/translate/preview_executor.py

@@ -18,7 +18,7 @@ from ...core.config_manager import ConfigManager
 from ...core.logger import belief_scope, logger
 from ...core.superset_client import SupersetClient
 from ...models.translate import TranslationJob
-from .preview_llm_client import LLMClient
+from ._llm_async_http import call_openai_compatible
 from .preview_resolve_provider import resolve_provider_model as _resolve_provider_model
 from .preview_response_parser import (
     compute_config_hash as _compute_config_hash,
@@ -38,7 +38,7 @@ class PreviewExecutor:
     # region fetch_sample_rows [TYPE Function]
     # @PURPOSE: Fetch sample rows from Superset dataset for preview.
     # @SIDE_EFFECT  Calls Superset chart data endpoint.
-    def fetch_sample_rows(
+    async def fetch_sample_rows(
         self,
         job: TranslationJob,
         sample_size: int = 10,
@@ -55,7 +55,7 @@ class PreviewExecutor:
                     raise ValueError("No Superset environments configured")
 
             client = SupersetClient(env_config)
-            dataset_detail = client.get_dataset_detail(int(job.source_datasource_id))
+            dataset_detail = await client.get_dataset_detail(int(job.source_datasource_id))
             query_context = client.build_dataset_preview_query_context(
                 dataset_id=int(job.source_datasource_id), dataset_record=dataset_detail,
                 template_params={}, effective_filters=[],
@@ -71,7 +71,7 @@ class PreviewExecutor:
             form_data.pop("query_mode", None)
 
             try:
-                response = client.network.request(
+                response = await client.network.request(
                     method="POST", endpoint="/api/v1/chart/data",
                     data=json.dumps(query_context),
                     headers={"Content-Type": "application/json"},
@@ -85,7 +85,7 @@ class PreviewExecutor:
     # region call_llm [TYPE Function]
     # @PURPOSE: Call the configured LLM provider with a prompt.
     # @SIDE_EFFECT  Makes HTTP call to LLM provider.
-    def call_llm(self, job: TranslationJob, prompt: str, max_tokens: int = 8192) -> str:
+    async def call_llm(self, job: TranslationJob, prompt: str, max_tokens: int = 8192) -> str:
         with belief_scope("PreviewExecutor.call_llm"):
             if not job.provider_id:
                 raise ValueError("Job has no LLM provider configured")
@@ -108,7 +108,7 @@ class PreviewExecutor:
             max_attempts = 2
             for attempt in range(max_attempts):
                 try:
-                    response_text = LLMClient.call_openai_compatible(
+                    content_text, _ = await call_openai_compatible(
                         base_url=provider.base_url, api_key=api_key, model=model,
                         prompt=prompt, provider_type=provider_type,
                         max_tokens=max_tokens * (attempt + 1),
@@ -121,7 +121,7 @@ class PreviewExecutor:
                         continue
                     raise
 
-            return response_text
+            return content_text
     # endregion call_llm
 
     # region delegation_helpers [TYPE Function]

backend/src/plugins/translate/superset_executor.py

@@ -9,8 +9,8 @@
 # @RATIONALE Direct SQL Lab API submission provides audit trail and execution monitoring within Superset.
 # @REJECTED Direct database connection bypass — would skip Superset's SQL Lab audit and RBAC.
 
+import asyncio
 import json
-import time
 from typing import Any
 import uuid
 
@@ -146,7 +146,7 @@ class SupersetSqlLabExecutor:
     # @PRE sql is valid SQL string. database_id is a valid Superset DB ID.
     # @POST Returns execution result dict with query_id and status.
     # @SIDE_EFFECT Makes HTTP POST to /api/v1/sqllab/execute/.
-    def execute_sql(
+    async def execute_sql(
         self,
         sql: str,
         database_id: int | None = None,
@@ -180,7 +180,7 @@ class SupersetSqlLabExecutor:
             # fetch() calls in Superset SQL Lab.
             endpoint = "/api/v1/sqllab/execute/"
             try:
-                response = client.network.request(
+                response = await client.network.request(
                     method="POST",
                     endpoint=endpoint,
                     data=json.dumps(payload),
@@ -248,7 +248,7 @@ class SupersetSqlLabExecutor:
     # @PRE query_id is a valid Superset query ID.
     # @POST Returns final execution status dict.
     # @SIDE_EFFECT Makes HTTP GET requests to Superset.
-    def poll_execution_status(
+    async def poll_execution_status(
         self,
         query_id: str,
         max_polls: int = 60,
@@ -265,7 +265,7 @@ class SupersetSqlLabExecutor:
 
             for attempt in range(max_polls):
                 try:
-                    response = client.network.request(
+                    response = await client.network.request(
                         method="GET",
                         endpoint=f"/api/v1/query/{query_id}",
                     )
@@ -308,11 +308,11 @@ class SupersetSqlLabExecutor:
                         }
 
                     if state in ("pending", "running", "started"):
-                        time.sleep(poll_interval_seconds)
+                        await asyncio.sleep(poll_interval_seconds)
                         continue
 
                     # Unknown state — treat as still running
-                    time.sleep(poll_interval_seconds)
+                    await asyncio.sleep(poll_interval_seconds)
 
                 except Exception as e:
                     logger.explore("Polling error, retrying", {
@@ -320,7 +320,7 @@ class SupersetSqlLabExecutor:
                         "attempt": attempt + 1,
                         "error": str(e),
                     })
-                    time.sleep(poll_interval_seconds)
+                    await asyncio.sleep(poll_interval_seconds)
                     continue
 
             # Timeout
@@ -341,7 +341,7 @@ class SupersetSqlLabExecutor:
     # @PRE sql is valid SQL.
     # @POST Returns final execution result.
     # @SIDE_EFFECT Makes HTTP calls to Superset API.
-    def execute_and_poll(
+    async def execute_and_poll(
         self,
         sql: str,
         database_id: int | None = None,
@@ -349,7 +349,7 @@ class SupersetSqlLabExecutor:
         poll_interval_seconds: float = 2.0,
     ) -> dict[str, Any]:
         with belief_scope("SupersetSqlLabExecutor.execute_and_poll"):
-            exec_result = self.execute_sql(
+            exec_result = await self.execute_sql(
                 sql=sql,
                 database_id=database_id,
                 run_async=False,  # Sync mode — Superset returns result directly
@@ -384,7 +384,7 @@ class SupersetSqlLabExecutor:
                     "query_id": None,
                 }
 
-            return self.poll_execution_status(
+            return await self.poll_execution_status(
                 query_id=str(query_id),
                 max_polls=max_polls,
                 poll_interval_seconds=poll_interval_seconds,
@@ -396,11 +396,11 @@ class SupersetSqlLabExecutor:
     # @PRE query_id is a valid Superset query ID.
     # @POST Returns query results if available.
     # @SIDE_EFFECT Makes HTTP GET to Superset.
-    def get_query_results(self, query_id: str) -> dict[str, Any]:
+    async def get_query_results(self, query_id: str) -> dict[str, Any]:
         with belief_scope("SupersetSqlLabExecutor.get_query_results"):
             client = self._get_client()
             try:
-                response = client.network.request(
+                response = await client.network.request(
                     method="GET",
                     endpoint=f"/api/v1/query/{query_id}/results",
                 )