ss-tools/backend/src/plugins/translate/orchestrator.py

# #region TranslationOrchestrator [C:5] [TYPE Module] [SEMANTICS sqlalchemy, tenacity, translate, orchestration, batch]
# @BRIEF Run lifecycle coordination: validate preconditions, dispatch executor, generate SQL, submit to Superset, record events.
# @LAYER: Domain
# @RELATION DEPENDS_ON -> [TranslationRun]
# @RELATION DEPENDS_ON -> [TranslationJob]
# @RELATION DEPENDS_ON -> [TranslationExecutor]
# @RELATION DEPENDS_ON -> [SQLGenerator]
# @RELATION DEPENDS_ON -> [SupersetSqlLabExecutor]
# @RELATION DEPENDS_ON -> [TranslationEventLog]
# @RELATION DEPENDS_ON -> [TranslationPreviewSession]
# @PRE: Valid job and accepted preview (for manual runs). Superset and LLM are reachable.
# @POST: Translation run is executed, SQL generated and submitted, events recorded.
# @SIDE_EFFECT: Creates TranslationRun; creates batches and records; generates SQL; submits to Superset; records events.
# @DATA_CONTRACT: Input[db, config_manager, current_user] -> Output[TranslationRun]
# @INVARIANT: State transitions: PENDING -> RUNNING -> COMPLETED|FAILED|CANCELLED. Exactly one terminal event per run.
# @RATIONALE: C5 orchestrator because preview, execution, event logging, and retry share state within single transaction boundary.
# @REJECTED: Distributed actor model (Celery) — eventual-consistency challenges at current scale.
# @REJECTED: stdout-only logging — lacks structured payload integrity; cannot enforce terminal-event invariant.

import json
import uuid
from collections.abc import Callable
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session, joinedload, selectinload

from ...core.config_manager import ConfigManager
from ...core.logger import belief_scope, logger
from ...models.translate import (
    TranslationBatch,
    TranslationJob,
    TranslationLanguage,
    TranslationPreviewSession,
    TranslationRecord,
    TranslationRun,
    TranslationRunLanguageStats,
)
from .events import TranslationEventLog
from .executor import TranslationExecutor
from .sql_generator import SQLGenerator, _normalize_timestamp_value
from .superset_executor import SupersetSqlLabExecutor


# #region TranslationOrchestrator [C:5] [TYPE Class]
# @BRIEF Coordinates full translation run lifecycle: validation, execution, SQL generation, Superset submission, event logging.
# @PRE: DB session and config manager are available.
# @POST: Runs are created, executed, and finalized with event records.
# @SIDE_EFFECT: Creates/modifies TranslationRun, TranslationBatch, TranslationRecord, TranslationEvent rows.
# @INVARIANT: State transitions: PENDING -> RUNNING -> COMPLETED|FAILED|CANCELLED.
class TranslationOrchestrator:

    def __init__(
        self,
        db: Session,
        config_manager: ConfigManager,
        current_user: str | None = None,
    ):
        self.db = db
        self.config_manager = config_manager
        self.current_user = current_user
        self.event_log = TranslationEventLog(db)
        self._job: TranslationJob | None = None

    # region start_run [TYPE Function]
    # @PURPOSE: Start a new translation run for a job.
    # @PRE: job_id exists. For manual runs, there must be an accepted preview session.
    # @POST: TranslationRun is created in PENDING status with hash fields and config snapshot. Events are recorded.
    # @SIDE_EFFECT: DB writes.
    def start_run(
        self,
        job_id: str,
        is_scheduled: bool = False,
        trigger_type: str | None = None,
        full_translation: bool = False,
    ) -> TranslationRun:
        with belief_scope("TranslationOrchestrator.start_run"):
            # Load and validate job
            job = self.db.query(TranslationJob).filter(TranslationJob.id == job_id).first()
            if not job:
                raise ValueError(f"Translation job '{job_id}' not found")
            self._job = job

            logger.reason("Starting translation run", {
                "job_id": job_id,
                "is_scheduled": is_scheduled,
            })

            # Validate preconditions
            self._validate_preconditions(job, is_scheduled=is_scheduled)

            # Compute hashes
            config_hash = self._compute_config_hash(job)
            dict_snapshot_hash = self._compute_dict_snapshot_hash(job_id)

            # Build config snapshot
            config_snapshot = {
                "source_dialect": job.source_dialect,
                "target_dialect": job.target_dialect,
                "database_dialect": job.database_dialect,
                "source_datasource_id": job.source_datasource_id,
                "source_table": job.source_table,
                "target_schema": job.target_schema,
                "target_table": job.target_table,
                "source_key_cols": job.source_key_cols,
                "target_key_cols": job.target_key_cols,
                "translation_column": job.translation_column,
                "target_column": job.target_column,
                "context_columns": job.context_columns,
                "provider_id": job.provider_id,
                "batch_size": job.batch_size,
                "upsert_strategy": job.upsert_strategy,
                "dictionary_ids": self._compute_dict_snapshot_hash(job_id),
                "full_translation": full_translation,
            }

            # Compute key_hash from source_key_cols
            import hashlib
            key_hash_input = json.dumps({
                "source_key_cols": job.source_key_cols,
                "source_datasource_id": job.source_datasource_id,
                "source_table": job.source_table,
            }, sort_keys=True)
            key_hash = hashlib.sha256(key_hash_input.encode()).hexdigest()[:16]

            # Create run record with all hash/snapshot fields
            run = TranslationRun(
                id=str(uuid.uuid4()),
                job_id=job_id,
                status="PENDING",
                trigger_type=trigger_type or ("scheduled" if is_scheduled else "manual"),
                config_snapshot=config_snapshot,
                key_hash=key_hash,
                config_hash=config_hash,
                dict_snapshot_hash=dict_snapshot_hash,
                created_by=self.current_user,
                created_at=datetime.now(UTC),
            )
            self.db.add(run)
            self.db.flush()

            # Record event
            self.event_log.log_event(
                job_id=job_id,
                run_id=run.id,
                event_type="RUN_STARTED",
                payload={
                    "is_scheduled": is_scheduled,
                    "trigger_type": run.trigger_type,
                    "config_hash": config_hash,
                    "dict_snapshot_hash": dict_snapshot_hash,
                    "key_hash": key_hash,
                },
                created_by=self.current_user,
            )

            self.db.commit()
            self.db.refresh(run)

            logger.reflect("Run created", {
                "run_id": run.id,
                "job_id": job_id,
                "status": run.status,
                "trigger_type": run.trigger_type,
            })
            return run
    # endregion start_run

    # region execute_run [TYPE Function]
    # @PURPOSE: Execute a translation run: dispatch executor, generate SQL, submit to Superset.
    # @PRE: run is in PENDING status.
    # @POST: Run is executed, SQL generated, Superset submission attempted.
    # @SIDE_EFFECT: LLM calls, DB writes, Superset API calls.
    def execute_run(
        self,
        run: TranslationRun,
        on_batch_progress: Callable[[str, int, int, int, int], None] | None = None,
        skip_insert: bool = False,
    ) -> TranslationRun:
        with belief_scope("TranslationOrchestrator.execute_run"):
            job = self.db.query(TranslationJob).filter(TranslationJob.id == run.job_id).first()
            if not job:
                raise ValueError(f"Job '{run.job_id}' not found")
            self._job = job

            if run.status != "PENDING":
                raise ValueError(
                    f"Cannot execute run in status '{run.status}'. "
                    f"Run must be in PENDING status."
                )

            logger.reason("Executing run", {
                "run_id": run.id,
                "job_id": job.id,
                "skip_insert": skip_insert,
            })

            # Record translation phase start
            self.event_log.log_event(
                job_id=job.id,
                run_id=run.id,
                event_type="TRANSLATION_PHASE_STARTED",
                payload={},
                created_by=self.current_user,
            )

            # Initialize per-language stats
            target_languages = job.target_languages or [job.target_dialect or "en"]
            if not isinstance(target_languages, list):
                target_languages = [str(target_languages)]
            language_stats_map: dict[str, TranslationRunLanguageStats] = {}
            for lang_code in target_languages:
                lang_stat = TranslationRunLanguageStats(
                    id=str(uuid.uuid4()),
                    run_id=run.id,
                    language_code=lang_code,
                    total_rows=0,
                    translated_rows=0,
                    failed_rows=0,
                    skipped_rows=0,
                    token_count=0,
                    estimated_cost=0.0,
                )
                self.db.add(lang_stat)
                language_stats_map[lang_code] = lang_stat
            self.db.flush()

            # Dispatch executor
            executor = TranslationExecutor(
                self.db, self.config_manager, self.current_user,
                on_batch_progress=on_batch_progress,
            )
            try:
                run = executor.execute_run(run, llm_progress_callback=None, language_stats_map=language_stats_map)
            except Exception as e:
                logger.explore("Translation execution failed", {
                    "run_id": run.id,
                    "error": str(e),
                })
                run.status = "FAILED"
                run.error_message = f"Translation execution failed: {e}"
                run.completed_at = datetime.now(UTC)
                self.db.flush()
                self.event_log.log_event(
                    job_id=job.id,
                    run_id=run.id,
                    event_type="RUN_FAILED",
                    payload={"error": str(e), "phase": "translation"},
                    created_by=self.current_user,
                )
                self.db.commit()
                return run

            # Check if executor cancelled itself due to cancellation flag
            if run.status == "CANCELLED":
                self._update_language_stats(run.id, language_stats_map)
                self.event_log.log_event(
                    job_id=job.id if job else (self._job.id if self._job else run.job_id),
                    run_id=run.id,
                    event_type="RUN_CANCELLED",
                    payload={"reason": "cancellation_flag"},
                    created_by=self.current_user,
                )
                self.db.commit()
                logger.reflect("Run cancelled via cancellation flag", {
                    "run_id": run.id,
                })
                return run

            # Aggregate per-language statistics after executor completes
            self._update_language_stats(run.id, language_stats_map)

            # Record translation phase complete
            self.event_log.log_event(
                job_id=job.id,
                run_id=run.id,
                event_type="TRANSLATION_PHASE_COMPLETED",
                payload={
                    "total": run.total_records,
                    "successful": run.successful_records,
                    "failed": run.failed_records,
                    "skipped": run.skipped_records,
                },
                created_by=self.current_user,
            )

            # Skip insert phase if requested (e.g., for preview-only execution)
            if skip_insert:
                run.status = "COMPLETED"
                run.completed_at = datetime.now(UTC)
                self.db.flush()
                self.event_log.log_event(
                    job_id=job.id,
                    run_id=run.id,
                    event_type="RUN_COMPLETED",
                    payload={"skip_insert": True},
                    created_by=self.current_user,
                )
                self.db.commit()
                return run

            # Generate SQL and submit to Superset
            insert_result = self._generate_and_insert_sql(job, run)
            run.insert_status = insert_result.get("status")
            run.superset_execution_id = str(insert_result.get("query_id") or "")
            run.superset_execution_log = insert_result
            # Preserve the translation-phase status.  If the executor already
            # marked the run FAILED (all LLM calls failed) we do NOT upgrade it
            # to COMPLETED just because the insert phase ran.
            if run.status != "FAILED":
                run.status = "COMPLETED"
            if insert_result.get("error_message"):
                run.error_message = insert_result["error_message"]
            run.completed_at = datetime.now(UTC)
            self.db.flush()

            # Record terminal event
            terminal_event = "RUN_COMPLETED"
            self.event_log.log_event(
                job_id=job.id,
                run_id=run.id,
                event_type=terminal_event,
                payload={
                    "insert_status": insert_result.get("status"),
                    "query_id": insert_result.get("query_id"),
                    "rows_affected": insert_result.get("rows_affected"),
                    "total_records": run.total_records,
                    "successful": run.successful_records,
                    "failed": run.failed_records,
                    "skipped": run.skipped_records,
                },
                created_by=self.current_user,
            )

            self.db.commit()
            # Re-query run after commit — refresh may fail if the object was
            # created in a different session or became detached during commit.
            run = self.db.query(TranslationRun).filter(TranslationRun.id == run.id).first()

            logger.reflect("Run execution complete", {
                "run_id": run.id,
                "status": run.status,
                "insert_status": run.insert_status,
            })
            return run
    # endregion execute_run

    # region _generate_and_insert_sql [TYPE Function]
    # @PURPOSE: Generate INSERT SQL from successful records and submit to Superset SQL Lab.
    # @PRE: job has target table configured. run has successful records.
    # @POST: SQL is generated and submitted. Returns execution result.
    # @SIDE_EFFECT: Superset API call.
    def _generate_and_insert_sql(
        self,
        job: TranslationJob,
        run: TranslationRun,
    ) -> dict[str, Any]:
        with belief_scope("TranslationOrchestrator._generate_and_insert_sql"):
            # Fetch successful records with eager-loaded language data
            records = (
                self.db.query(TranslationRecord)
                .options(joinedload(TranslationRecord.languages))
                .filter(
                    TranslationRecord.run_id == run.id,
                    TranslationRecord.status == "SUCCESS",
                    TranslationRecord.target_sql.isnot(None),
                )
                .all()
            )

            if not records:
                logger.reason("No successful records to insert", {"run_id": run.id})
                return {"status": "skipped", "reason": "no_records", "query_id": None}

            logger.reason(f"Generating SQL for {len(records)} records", {
                "run_id": run.id,
                "dialect": job.database_dialect or job.target_dialect,
            })

            effective_target = job.target_column or job.translation_column
            primary_language = (job.target_languages or ["en"])[0]

            # Columns that exist in the target ClickHouse table
            columns = []
            if job.target_key_cols:
                columns.extend(job.target_key_cols)
            if effective_target:
                columns.append(effective_target)
            if job.target_language_column:
                columns.append(job.target_language_column)
            if job.target_source_column:
                columns.append(job.target_source_column)
            if job.target_source_language_column:
                columns.append(job.target_source_language_column)
            columns.append("context")
            columns.append("is_original")
            # Deduplicate while preserving order
            seen: set[str] = set()
            deduped: list[str] = []
            for c in columns:
                if c and c not in seen:
                    deduped.append(c)
                    seen.add(c)
            columns = deduped

            # Keys for the context JSON: context_columns + original translation_column
            context_keys = list(job.context_columns or [])
            if job.translation_column and job.translation_column != effective_target and job.translation_column not in context_keys:
                context_keys.append(job.translation_column)

            rows_for_sql: list[dict[str, object]] = []
            for rec in records:
                source_data = rec.source_data or {}

                # Detect source language from first TranslationLanguage entry
                detected_src_lang = "und"
                if rec.languages and len(rec.languages) > 0:
                    detected_src_lang = rec.languages[0].source_language_detected or "und"

                # Build context JSON: all extra columns the user configured
                context_data: dict[str, str] = {}
                for key in context_keys:
                    val = source_data.get(key)
                    context_data[key] = str(val) if val is not None else ""

                # ── Shared base row ──
                base_row: dict[str, object] = {}

                if job.target_key_cols:
                    for k in job.target_key_cols:
                        raw = source_data.get(k)
                        if raw is not None:
                            normalized = _normalize_timestamp_value(raw)
                            base_row[k] = normalized if normalized else raw
                        else:
                            base_row[k] = None

                if job.target_source_column:
                    base_row[job.target_source_column] = rec.source_sql or ""

                if job.target_source_language_column:
                    base_row[job.target_source_language_column] = detected_src_lang

                base_row["context"] = json.dumps(context_data, ensure_ascii=False)

                # ── 1. ORIGINAL row (is_original = 1) ──
                original_row = dict(base_row)
                if effective_target:
                    original_row[effective_target] = rec.source_sql or ""
                if job.target_language_column:
                    original_row[job.target_language_column] = detected_src_lang
                original_row["is_original"] = 1
                rows_for_sql.append(original_row)

                # ── 2. TRANSLATION rows (is_original = 0) ──
                # Skip language that matches the source — the original row already covers it
                if rec.languages and len(rec.languages) > 0:
                    for lang in rec.languages:
                        if lang.language_code == detected_src_lang:
                            continue
                        trans_row = dict(base_row)
                        trans_value = lang.final_value or lang.translated_value or ""
                        if effective_target:
                            trans_row[effective_target] = trans_value
                        if job.target_language_column:
                            trans_row[job.target_language_column] = lang.language_code
                        trans_row["is_original"] = 0
                        rows_for_sql.append(trans_row)
                else:
                    # Fallback: no per-language data
                    fallback_row = dict(base_row)
                    if effective_target:
                        fallback_row[effective_target] = rec.target_sql or ""
                    if job.target_language_column:
                        fallback_row[job.target_language_column] = primary_language
                    fallback_row["is_original"] = 0
                    rows_for_sql.append(fallback_row)

            if not columns:
                columns = [effective_target or "translated_text"]
                rows_for_sql = [{columns[0]: rec.target_sql or ""} for rec in records]

            # Resolve the real database backend engine from Superset
            try:
                env_id = job.environment_id or job.source_dialect or ""
                executor = SupersetSqlLabExecutor(self.config_manager, env_id)
                executor.resolve_database_id(
                    target_database_id=job.target_database_id,
                )
                real_backend = executor.get_database_backend()
            except Exception as e:
                logger.explore("Failed to resolve database backend, falling back to job dialet", {
                    "error": str(e),
                })
                real_backend = None

            dialect = real_backend or job.database_dialect or job.target_dialect or "postgresql"

            # Generate SQL
            try:
                sql, row_count = SQLGenerator.generate(
                    dialect=dialect,
                    target_schema=job.target_schema,
                    target_table=job.target_table or "translated_data",
                    columns=columns,
                    rows=rows_for_sql,
                    key_columns=job.target_key_cols,
                    upsert_strategy=job.upsert_strategy or "MERGE",
                )
            except ValueError as e:
                logger.explore("SQL generation failed", {"error": str(e)})
                return {"status": "failed", "error_message": str(e), "query_id": None}

            logger.reason("SQL generated with dialect", {
                "dialect": dialect,
                "real_backend": real_backend,
                "job_database_dialect": job.database_dialect,
                "job_target_dialect": job.target_dialect,
            })

            # Log insert phase start
            self.event_log.log_event(
                job_id=job.id,
                run_id=run.id,
                event_type="INSERT_PHASE_STARTED",
                payload={"sql_length": len(sql), "row_count": row_count, "dialect": dialect},
                created_by=self.current_user,
            )

            # Submit to Superset
            try:
                result = executor.execute_and_poll(
                    sql=sql,
                    max_polls=30,
                    poll_interval_seconds=2.0,
                )
            except Exception as e:
                logger.explore("Superset SQL submission failed", {
                    "run_id": run.id,
                    "error": str(e),
                })
                result = {"status": "failed", "error_message": str(e), "query_id": None}

            # Log insert phase complete
            self.event_log.log_event(
                job_id=job.id,
                run_id=run.id,
                event_type="INSERT_PHASE_COMPLETED",
                payload=result,
                created_by=self.current_user,
            )

            return result
    # endregion _generate_and_insert_sql

    # region _validate_preconditions [TYPE Function]
    # @PURPOSE: Validate preconditions before starting a run.
    # @PRE: None.
    # @POST: Raises ValueError if preconditions are not met.
    # @SIDE_EFFECT: None.
    def _validate_preconditions(
        self,
        job: TranslationJob,
        is_scheduled: bool = False,
    ) -> None:
        with belief_scope("TranslationOrchestrator._validate_preconditions"):
            # Job must be in valid status
            if job.status in ("DRAFT",):
                raise ValueError(
                    f"Cannot run job '{job.id}' in status '{job.status}'. "
                    f"Job must be READY, ACTIVE, or COMPLETED."
                )

            # Must have target table configured
            if not job.target_table:
                raise ValueError(
                    f"Job '{job.id}' has no target table configured. "
                    "Configure a target table before running."
                )

            # Must have LLM provider configured
            if not job.provider_id:
                raise ValueError(
                    f"Job '{job.id}' has no LLM provider configured. "
                    "Select an LLM provider before running."
                )

            # Must have a translation column
            if not job.translation_column:
                raise ValueError(
                    f"Job '{job.id}' has no translation column configured. "
                    "Select a translation column before running."
                )

            # For manual runs, must have accepted preview
            if not is_scheduled:
                accepted_session = (
                    self.db.query(TranslationPreviewSession)
                    .filter(
                        TranslationPreviewSession.job_id == job.id,
                        TranslationPreviewSession.status == "APPLIED",
                    )
                    .order_by(TranslationPreviewSession.created_at.desc())
                    .first()
                )
                if not accepted_session:
                    raise ValueError(
                        f"Job '{job.id}' has no accepted preview session. "
                        "Run and accept a preview before executing a manual translation run."
                    )

            logger.reason("Preconditions validated", {
                "job_id": job.id,
                "is_scheduled": is_scheduled,
            })
    # endregion _validate_preconditions

    # region retry_failed_batches [TYPE Function]
    # @PURPOSE: Retry failed batches in a run.
    # @PRE: run exists and has failed batches.
    # @POST: Failed batches are re-processed.
    # @SIDE_EFFECT: LLM calls, DB writes.
    def retry_failed_batches(
        self,
        run_id: str,
    ) -> TranslationRun:
        with belief_scope("TranslationOrchestrator.retry_failed_batches"):
            run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
            if not run:
                raise ValueError(f"Run '{run_id}' not found")

            job = self.db.query(TranslationJob).filter(TranslationJob.id == run.job_id).first()
            if not job:
                raise ValueError(f"Job '{run.job_id}' not found")
            self._job = job

            # Find failed batches
            failed_batches = (
                self.db.query(TranslationBatch)
                .filter(
                    TranslationBatch.run_id == run_id,
                    TranslationBatch.status.in_(["FAILED", "COMPLETED_WITH_ERRORS"]),
                )
                .all()
            )

            if not failed_batches:
                raise ValueError(f"No failed batches found for run '{run_id}'")

            logger.reason("Retrying failed batches", {
                "run_id": run_id,
                "batch_count": len(failed_batches),
            })

            self.event_log.log_event(
                job_id=job.id,
                run_id=run.id,
                event_type="RUN_RETRYING",
                payload={"batch_count": len(failed_batches)},
                created_by=self.current_user,
            )

            # Re-process each failed batch
            executor = TranslationExecutor(self.db, self.config_manager, self.current_user)
            run.status = "RUNNING"
            self.db.flush()

            for batch in failed_batches:
                # Fetch failed records for this batch
                failed_records = (
                    self.db.query(TranslationRecord)
                    .filter(
                        TranslationRecord.batch_id == batch.id,
                        TranslationRecord.status == "FAILED",
                    )
                    .all()
                )

                if not failed_records:
                    continue

                # Build rows from failed records
                retry_rows = []
                for rec in failed_records:
                    retry_rows.append({
                        "row_index": rec.source_object_id or "0",
                        "source_text": rec.source_sql or "",
                        "approved_translation": None,
                        "source_object_name": rec.source_object_name or "",
                    })

                # Process retry batch
                result = executor._process_batch(
                    job=job,
                    run_id=run_id,
                    batch_index=batch.batch_index,
                    batch_rows=retry_rows,
                )

                # Update run stats
                run.successful_records = (run.successful_records or 0) + result["successful"]
                run.failed_records = (run.failed_records or 0) + result["failed"]
                run.skipped_records = (run.skipped_records or 0) + result["skipped"]
                self.db.flush()

            # Update run status
            if run.failed_records == 0:
                run.status = "COMPLETED"
            elif run.successful_records == 0:
                run.status = "FAILED"
            else:
                run.status = "COMPLETED"

            run.completed_at = datetime.now(UTC)
            self.db.flush()

            self.event_log.log_event(
                job_id=job.id,
                run_id=run.id,
                event_type="RUN_COMPLETED",
                payload={"retry": True},
                created_by=self.current_user,
            )

            self.db.commit()
            logger.reflect("Retry complete", {
                "run_id": run_id,
                "status": run.status,
            })
            return run
    # endregion retry_failed_batches

    # region retry_insert [TYPE Function]
    # @PURPOSE: Retry the SQL insert phase for a completed run.
    # @PRE: run exists and has successful records.
    # @POST: SQL is regenerated and re-submitted to Superset.
    # @SIDE_EFFECT: Superset API call.
    def retry_insert(
        self,
        run_id: str,
    ) -> TranslationRun:
        with belief_scope("TranslationOrchestrator.retry_insert"):
            run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
            if not run:
                raise ValueError(f"Run '{run_id}' not found")

            job = self.db.query(TranslationJob).filter(TranslationJob.id == run.job_id).first()
            if not job:
                raise ValueError(f"Job '{run.job_id}' not found")
            self._job = job

            logger.reason("Retrying insert phase", {
                "run_id": run_id,
            })

            self.event_log.log_event(
                job_id=job.id,
                run_id=run.id,
                event_type="RUN_RETRY_INSERT",
                payload={},
                created_by=self.current_user,
            )

            # Regenerate SQL and submit
            insert_result = self._generate_and_insert_sql(job, run)

            # Update run
            run.insert_status = insert_result.get("status")
            run.superset_execution_id = str(insert_result.get("query_id") or "")
            if insert_result.get("error_message"):
                run.error_message = insert_result["error_message"]
            self.db.flush()
            self.db.commit()
            self.db.refresh(run)

            logger.reflect("Insert retry complete", {
                "run_id": run_id,
                "insert_status": run.insert_status,
            })
            return run
    # endregion retry_insert

    # region cancel_run [TYPE Function]
    # @PURPOSE: Cancel a running translation.
    # @PRE: run is in PENDING or RUNNING status.
    # @POST: Run status is set to CANCELLED.
    # @SIDE_EFFECT: DB write; records event.
    def cancel_run(self, run_id: str) -> TranslationRun:
        with belief_scope("TranslationOrchestrator.cancel_run"):
            # Set short lock timeout to avoid blocking on row lock held by
            # background executor thread (which holds RowExclusiveLock during
            # batch processing). If the row is locked, we fall back to setting
            # a cancellation flag via direct SQL UPDATE, which the executor
            # checks after each batch commit.
            try:
                self.db.execute(text("SET LOCAL lock_timeout = '3s'"))
                run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
            except Exception:
                # Row is locked — set cancellation flag via direct SQL
                logger.explore("Row lock timeout — setting cancellation flag via direct SQL", {
                    "run_id": run_id,
                })
                self.db.execute(
                    text("UPDATE translation_runs SET error_message = 'CANCEL_REQUESTED' WHERE id = :run_id"),
                    {"run_id": run_id},
                )
                self.db.commit()
                run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
                if not run:
                    raise ValueError(f"Run '{run_id}' not found")
                logger.reflect("Cancellation flag set for locked run", {
                    "run_id": run_id,
                })
                return run

            if not run:
                raise ValueError(f"Run '{run_id}' not found")

            if run.status not in ("PENDING", "RUNNING"):
                raise ValueError(
                    f"Cannot cancel run in status '{run.status}'. "
                    f"Only PENDING or RUNNING runs can be cancelled."
                )

            run.status = "CANCELLED"
            run.completed_at = datetime.now(UTC)
            self.db.flush()

            self.event_log.log_event(
                job_id=run.job_id,
                run_id=run.id,
                event_type="RUN_CANCELLED",
                payload={},
                created_by=self.current_user,
            )

            self.db.commit()
            self.db.refresh(run)

            logger.reflect("Run cancelled", {
                "run_id": run_id,
            })
            return run
    # endregion cancel_run

    # region get_run_status [TYPE Function]
    # @PURPOSE: Get run status with statistics.
    # @PRE: run_id exists.
    # @POST: Returns dict with run details.
    def get_run_status(self, run_id: str) -> dict[str, Any]:
        with belief_scope("TranslationOrchestrator.get_run_status"):
            run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
            if not run:
                raise ValueError(f"Run '{run_id}' not found")

            # Count batches
            batch_count = (
                self.db.query(TranslationBatch)
                .filter(TranslationBatch.run_id == run_id)
                .count()
            )

            # Get event summary
            event_summary = self.event_log.get_run_event_summary(run_id)

            # Get language stats
            language_stats_entries = (
                self.db.query(TranslationRunLanguageStats)
                .filter(TranslationRunLanguageStats.run_id == run_id)
                .all()
            )
            language_stats = [
                {
                    "language_code": ls.language_code,
                    "total_rows": ls.total_rows or 0,
                    "translated_rows": ls.translated_rows or 0,
                    "failed_rows": ls.failed_rows or 0,
                    "skipped_rows": ls.skipped_rows or 0,
                    "token_count": ls.token_count or 0,
                    "estimated_cost": ls.estimated_cost or 0.0,
                }
                for ls in language_stats_entries
            ]

            return {
                "id": run.id,
                "job_id": run.job_id,
                "status": run.status,
                "started_at": run.started_at.isoformat() if run.started_at else None,
                "completed_at": run.completed_at.isoformat() if run.completed_at else None,
                "error_message": run.error_message,
                "total_records": run.total_records or 0,
                "successful_records": run.successful_records or 0,
                "failed_records": run.failed_records or 0,
                "skipped_records": run.skipped_records or 0,
                "insert_status": run.insert_status,
                "superset_execution_id": run.superset_execution_id,
                "batch_count": batch_count,
                "language_stats": language_stats,
                "event_invariants": {
                    "has_run_started": event_summary["has_run_started"],
                    "terminal_event_count": event_summary["terminal_event_count"],
                    "invariant_valid": event_summary["invariant_valid"],
                },
                "created_by": run.created_by,
                "created_at": run.created_at.isoformat() if run.created_at else None,
            }
    # endregion get_run_status

    # region get_run_records [TYPE Function]
    # @PURPOSE: Get paginated records for a run.
    # @PRE: run_id exists.
    # @POST: Returns dict with records and pagination info.
    def get_run_records(
        self,
        run_id: str,
        page: int = 1,
        page_size: int = 50,
        status_filter: str | None = None,
    ) -> dict[str, Any]:
        with belief_scope("TranslationOrchestrator.get_run_records"):
            query = (
                self.db.query(TranslationRecord)
                .options(selectinload(TranslationRecord.languages))
                .filter(TranslationRecord.run_id == run_id)
            )

            if status_filter:
                query = query.filter(TranslationRecord.status == status_filter)

            total = query.count()
            offset = (page - 1) * page_size
            records = (
                query.order_by(TranslationRecord.created_at.desc())
                .offset(offset)
                .limit(page_size)
                .all()
            )

            return {
                "items": [
                    {
                        "id": r.id,
                        "batch_id": r.batch_id,
                        "source_sql": r.source_sql,
                        "target_sql": r.target_sql,
                        "source_object_type": r.source_object_type,
                        "source_object_id": r.source_object_id,
                        "source_object_name": r.source_object_name,
                        "status": r.status,
                        "error_message": r.error_message,
                        "created_at": r.created_at.isoformat() if r.created_at else None,
                        "languages": [
                            {
                                "language_code": tl.language_code,
                                "translated_value": tl.translated_value,
                                "final_value": tl.final_value or tl.translated_value,
                                "source_language_detected": tl.source_language_detected,
                                "status": tl.status,
                                "needs_review": tl.needs_review or False,
                            }
                            for tl in (r.languages or [])
                        ],
                    }
                    for r in records
                ],
                "total": total,
                "page": page,
                "page_size": page_size,
                "status_filter": status_filter,
            }
    # endregion get_run_records

    # region get_run_history [TYPE Function]
    # @PURPOSE: Get run history for a job.
    # @PRE: job_id exists.
    # @POST: Returns list of runs.
    def get_run_history(
        self,
        job_id: str,
        page: int = 1,
        page_size: int = 20,
    ) -> tuple[int, list[dict[str, Any]]]:
        with belief_scope("TranslationOrchestrator.get_run_history"):
            query = self.db.query(TranslationRun).filter(
                TranslationRun.job_id == job_id
            )
            total = query.count()
            runs = (
                query.order_by(TranslationRun.created_at.desc())
                .offset((page - 1) * page_size)
                .limit(page_size)
                .all()
            )

            return total, [
                {
                    "id": r.id,
                    "job_id": r.job_id,
                    "status": r.status,
                    "started_at": r.started_at.isoformat() if r.started_at else None,
                    "completed_at": r.completed_at.isoformat() if r.completed_at else None,
                    "error_message": r.error_message,
                    "total_records": r.total_records or 0,
                    "successful_records": r.successful_records or 0,
                    "failed_records": r.failed_records or 0,
                    "skipped_records": r.skipped_records or 0,
                    "insert_status": r.insert_status,
                    "superset_execution_id": r.superset_execution_id,
                    "created_by": r.created_by,
                    "created_at": r.created_at.isoformat() if r.created_at else None,
                }
                for r in runs
            ]
    # endregion get_run_history

    # region _update_language_stats [TYPE Function]
    # @PURPOSE: Aggregate TranslationLanguage entries by language_code and update TranslationRunLanguageStats.
    # @PRE: run_id and language_stats_map are valid. DB session is available.
    # @POST: Language stats are updated with row counts and estimated tokens/cost.
    # @SIDE_EFFECT: DB writes on language_stats objects.
    def _update_language_stats(
        self,
        run_id: str,
        language_stats_map: dict[str, TranslationRunLanguageStats],
    ) -> None:
        with belief_scope("TranslationOrchestrator._update_language_stats"):
            # Get all records for this run to join with TranslationLanguage
            records = (
                self.db.query(TranslationRecord)
                .filter(TranslationRecord.run_id == run_id)
                .all()
            )
            record_ids = [r.id for r in records]

            if not record_ids:
                logger.reason("No records for language stats aggregation", {"run_id": run_id})
                return

            # Get all language entries for this run's records
            lang_entries = (
                self.db.query(TranslationLanguage)
                .filter(TranslationLanguage.record_id.in_(record_ids))
                .all()
            )

            # Aggregate by language_code
            from collections import defaultdict
            agg: dict[str, dict[str, int]] = defaultdict(lambda: {"total": 0, "translated": 0, "failed": 0, "skipped": 0})

            for le in lang_entries:
                code = le.language_code
                agg[code]["total"] += 1
                if le.status in ("translated", "approved", "edited"):
                    agg[code]["translated"] += 1
                elif le.status == "failed":
                    agg[code]["failed"] += 1
                elif le.status == "skipped":
                    agg[code]["skipped"] += 1

            # Estimate tokens: heuristic based on character count of translated values
            total_chars = sum(
                len(le.translated_value or "") for le in lang_entries if le.translated_value
            )
            total_tokens = max(1, total_chars // 4)  # ~4 chars per token
            cost_per_token = 0.002 / 1000  # $0.002 per 1K tokens

            # Update each language stat entry
            for lang_code, lang_stat in language_stats_map.items():
                data = agg.get(lang_code, {"total": 0, "translated": 0, "failed": 0, "skipped": 0})
                lang_stat.total_rows = data["total"]
                lang_stat.translated_rows = data["translated"]
                lang_stat.failed_rows = data["failed"]
                lang_stat.skipped_rows = data["skipped"]

                # Proportional token split: share tokens across languages
                num_langs = len(language_stats_map)
                if num_langs > 0:
                    lang_stat.token_count = total_tokens // num_langs
                    lang_stat.estimated_cost = round((lang_stat.token_count / 1000) * cost_per_token, 6)

            self.db.flush()

            logger.reason("Language stats updated", {
                "run_id": run_id,
                "languages": list(language_stats_map.keys()),
                "total_tokens_est": total_tokens,
            })
    # endregion _update_language_stats

    # region _compute_config_hash [TYPE Function]
    # @PURPOSE: Compute a hash of the job's current configuration for snapshot comparison.
    @staticmethod
    def _compute_config_hash(job: TranslationJob) -> str:
        import hashlib
        config_str = json.dumps({
            "source_dialect": job.source_dialect,
            "target_dialect": job.target_dialect,
            "source_datasource_id": job.source_datasource_id,
            "translation_column": job.translation_column,
            "context_columns": job.context_columns,
            "target_languages": sorted(job.target_languages) if job.target_languages else [],
            "provider_id": job.provider_id,
            "batch_size": job.batch_size,
            "upsert_strategy": job.upsert_strategy,
        }, sort_keys=True)
        return hashlib.sha256(config_str.encode()).hexdigest()[:16]
    # endregion _compute_config_hash

    # region _compute_dict_snapshot_hash [TYPE Function]
    # @PURPOSE: Compute a hash of dictionary state for snapshot comparison.
    def _compute_dict_snapshot_hash(self, job_id: str) -> str:
        import hashlib

        from ...models.translate import TranslationJobDictionary
        dict_links = (
            self.db.query(TranslationJobDictionary)
            .filter(TranslationJobDictionary.job_id == job_id)
            .all()
        )
        dict_ids = sorted([dl.dictionary_id for dl in dict_links])
        hash_input = ",".join(dict_ids)
        return hashlib.sha256(hash_input.encode()).hexdigest()[:16]
    # endregion _compute_dict_snapshot_hash

# #endregion TranslationOrchestrator