# #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. from collections.abc import Callable from typing import Any from sqlalchemy.orm import Session from ...core.config_manager import ConfigManager from ...core.logger import belief_scope from ...models.translate import ( TranslationRun, ) from .events import TranslationEventLog from .orchestrator_aggregator import TranslationResultAggregator from .orchestrator_planner import TranslationPlanner from .orchestrator_runner import TranslationStageRunner # #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. # @RATIONALE Delegates to TranslationPlanner, TranslationStageRunner, TranslationResultAggregator for SOLID decomposition. # @REJECTED Monolithic class was rejected — violated INV_7 (single contract < 150 lines, module < 400 lines). class TranslationOrchestrator: """Coordinates full translation run lifecycle via delegated sub-components.""" 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._planner = TranslationPlanner(db, self.event_log, current_user) self._runner = TranslationStageRunner(db, config_manager, self.event_log, current_user) self._aggregator = TranslationResultAggregator(db, self.event_log) # region start_run [TYPE Function] # @PURPOSE: Start a new translation run for a job. # @PRE job_id exists. For manual runs, an accepted preview session must exist. # @POST TranslationRun is created in PENDING status with hash fields and config snapshot. # @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"): return self._planner.plan_run( job_id=job_id, is_scheduled=is_scheduled, trigger_type=trigger_type, full_translation=full_translation, ) # 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"): return self._runner.execute_run( run=run, on_batch_progress=on_batch_progress, skip_insert=skip_insert, ) # endregion execute_run # region retry_failed_batches [TYPE Function] # @PURPOSE: Retry failed batches in a run. def retry_failed_batches(self, run_id: str) -> TranslationRun: return self._runner.retry_failed_batches(run_id) # endregion retry_failed_batches # region retry_insert [TYPE Function] # @PURPOSE: Retry the SQL insert phase for a completed run. def retry_insert(self, run_id: str) -> TranslationRun: return self._runner.retry_insert(run_id) # endregion retry_insert # region cancel_run [TYPE Function] # @PURPOSE: Cancel a running translation. def cancel_run(self, run_id: str) -> TranslationRun: return self._runner.cancel_run(run_id) # endregion cancel_run # region _generate_and_insert_sql [TYPE Function] [SEMANTICS backward-compat wrapper] # @PURPOSE: Backward-compatible delegating wrapper for SQL generation and insert. # @SIDE_EFFECT Delegates to SQLInsertService. May call Superset API. def _generate_and_insert_sql( self, job: Any, run: TranslationRun, ) -> dict[str, Any]: """Backward-compatible wrapper — delegates to SQLInsertService.""" from .orchestrator_sql import SQLInsertService svc = SQLInsertService(self.db, self.config_manager, self.event_log) return svc.generate_and_insert_sql(job, run) # endregion _generate_and_insert_sql # region _update_language_stats [TYPE Function] [SEMANTICS backward-compat wrapper] # @PURPOSE: Backward-compatible delegating wrapper for language stats update. # @SIDE_EFFECT Delegates to TranslationResultAggregator.update_language_stats. def _update_language_stats( self, run_id: str, language_stats_map: dict[str, Any], ) -> None: """Backward-compatible wrapper — delegates to TranslationResultAggregator.""" return self._aggregator.update_language_stats(run_id, language_stats_map) # endregion _update_language_stats # region get_run_status [TYPE Function] # @PURPOSE: Get run status with statistics. def get_run_status(self, run_id: str) -> dict[str, Any]: return self._aggregator.get_run_status(run_id) # endregion get_run_status # region get_run_records [TYPE Function] # @PURPOSE: Get paginated records for a run. def get_run_records( self, run_id: str, page: int = 1, page_size: int = 50, status_filter: str | None = None, deduplicate: bool = False, ) -> dict[str, Any]: return self._aggregator.get_run_records(run_id, page, page_size, status_filter, deduplicate=deduplicate) # endregion get_run_records # region get_run_history [TYPE Function] # @PURPOSE: Get run history for a job. def get_run_history( self, job_id: str, page: int = 1, page_size: int = 20, ) -> tuple[int, list[dict[str, Any]]]: return self._aggregator.get_run_history(job_id, page, page_size) # endregion get_run_history # #endregion TranslationOrchestrator