software/src/orchestration/orchestrator.py

1"""Main orchestrator for fault-tolerant Bible translation.23Coordinates parallel book translation across multiple processes,4handling job state, graceful shutdown, and error recovery.5"""67from __future__ import annotations89import json10import logging11import os12import time13from concurrent.futures import ThreadPoolExecutor, as_completed14from datetime import datetime15from pathlib import Path16from typing import TYPE_CHECKING, Any1718from src.orchestration.models import (19    ALL_BOOKS,20    NEW_TESTAMENT_BOOKS,21    OLD_TESTAMENT_BOOKS,22    BookResult,23    JobResult,24    JobStatus,25)26from src.orchestration.provenance import (27    TranslationProvenance,28    compute_glossary_hash,29    compute_hash,30    compute_style_hash,31)32from src.orchestration.rate_limiter import DummyRateLimiter, SharedRateLimiter33from src.orchestration.signals import GracefulShutdown34from src.orchestration.state import JobStateStore3536if TYPE_CHECKING:37    from src.models.verse import VerseReference3839logger = logging.getLogger(__name__)404142class JobOrchestrator:43    """Orchestrates fault-tolerant Bible translation across multiple processes.4445    Features:46    - Parallel book translation using multiprocessing47    - Verse-level progress tracking with SQLite persistence48    - Graceful shutdown handling (Ctrl+C)49    - Resume capability from any point50    - Full provenance tracking5152    Example:53        orchestrator = JobOrchestrator(data_dir=Path("./data"))5455        # Start new job56        job_id = orchestrator.create_job(books=["john", "matthew"])57        result = orchestrator.run(job_id)5859        # Resume after crash60        result = orchestrator.resume()61    """6263    def __init__(64        self,65        data_dir: Path,66        max_parallel_books: int = 4,67        max_concurrent_api_calls: int = 5,68        rate_limit_delay: float = 1.2,69    ) -> None:70        """Initialize the job orchestrator.7172        Args:73            data_dir: Path to the data directory.74            max_parallel_books: Maximum books to translate in parallel.75            max_concurrent_api_calls: Maximum concurrent API calls across all workers.76            rate_limit_delay: Delay between API calls in seconds.77        """78        self.data_dir = Path(data_dir)79        self.max_parallel_books = max_parallel_books80        self.max_concurrent_api_calls = max_concurrent_api_calls81        self.rate_limit_delay = rate_limit_delay8283        # Initialize state store84        self.state_store = JobStateStore(self.data_dir / "jobs" / "state.db")8586        # Shutdown handler (set during run)87        self._shutdown: GracefulShutdown | None = None8889    def create_job(90        self,91        books: list[str] | None = None,92        nt_only: bool = False,93        ot_only: bool = False,94        skip_completed: bool = True,95        job_id: str | None = None,96    ) -> str:97        """Create a new Bible translation job.9899        Args:100            books: Specific books to translate (default: all 66).101            nt_only: Only translate New Testament (27 books).102            ot_only: Only translate Old Testament (39 books).103            skip_completed: Skip books that are already fully translated.104            job_id: Optional specific job ID.105106        Returns:107            Job ID for the created job.108        """109        # Determine which books to translate110        if books:111            selected_books = [b.lower() for b in books]112        elif nt_only:113            selected_books = list(NEW_TESTAMENT_BOOKS)114        elif ot_only:115            selected_books = list(OLD_TESTAMENT_BOOKS)116        else:117            selected_books = list(ALL_BOOKS)118119        # Validate book names120        invalid = set(selected_books) - set(ALL_BOOKS)121        if invalid:122            raise ValueError(f"Invalid book names: {invalid}")123124        # Load current configuration125        config = self._snapshot_config()126127        # Create job in state store128        job = self.state_store.create_bible_job(129            books=selected_books,130            config=config,131            job_id=job_id,132        )133134        logger.info("Created job %s with %d books", job.job_id, len(selected_books))135136        return job.job_id137138    def run(self, job_id: str) -> JobResult:139        """Run a Bible translation job.140141        Args:142            job_id: Job identifier to run.143144        Returns:145            JobResult with completion status and statistics.146        """147        job = self.state_store.get_bible_job(job_id)148        if not job:149            raise ValueError(f"Job not found: {job_id}")150151        if job.status == JobStatus.COMPLETED:152            raise ValueError(f"Job already completed: {job_id}")153154        # Mark job as started155        start_time = datetime.utcnow()156        self.state_store.update_job_status(157            job_id, JobStatus.IN_PROGRESS, started_at=start_time158        )159160        book_results: list[BookResult] = []161162        with GracefulShutdown() as shutdown:163            self._shutdown = shutdown164165            # Create thread-safe rate limiter166            rate_limiter = SharedRateLimiter(167                max_concurrent=self.max_concurrent_api_calls,168                delay=self.rate_limit_delay,169            )170171            try:172                # Initialize verse jobs for all books173                self._initialize_all_books(job_id, job.books)174175                # Process books in parallel using thread pool176                if self.max_parallel_books > 1:177                    book_results = self._run_parallel(178                        job_id, job.books, rate_limiter, shutdown179                    )180                else:181                    book_results = self._run_sequential(182                        job_id, job.books, rate_limiter, shutdown183                    )184185            finally:186                rate_limiter.shutdown()187188        # Compute final statistics189        end_time = datetime.utcnow()190        duration = (end_time - start_time).total_seconds()191192        books_completed = sum(1 for r in book_results if r.success)193        books_failed = len(book_results) - books_completed194        total_translated = sum(r.verses_translated for r in book_results)195        total_failed = sum(r.verses_failed for r in book_results)196197        # Determine final status198        all_success = books_failed == 0 and not shutdown.should_stop199        final_status = JobStatus.COMPLETED if all_success else JobStatus.FAILED200201        self.state_store.update_job_status(202            job_id, final_status, completed_at=end_time203        )204205        return JobResult(206            job_id=job_id,207            success=all_success,208            books_completed=books_completed,209            books_failed=books_failed,210            total_verses_translated=total_translated,211            total_verses_failed=total_failed,212            duration_seconds=duration,213            book_results=book_results,214        )215216    def resume(self, job_id: str | None = None) -> JobResult:217        """Resume an incomplete job.218219        Args:220            job_id: Optional specific job ID. If None, resumes latest incomplete.221222        Returns:223            JobResult with completion status.224        """225        if job_id:226            job = self.state_store.get_bible_job(job_id)227            if not job:228                raise ValueError(f"Job not found: {job_id}")229        else:230            job = self.state_store.get_latest_incomplete_job()231            if not job:232                raise ValueError("No incomplete jobs found")233234        logger.info("Resuming job %s", job.job_id)235236        # Reset any orphaned verses (in_progress but not completed)237        reset_count = self.state_store.reset_orphaned_verses(job.job_id)238        if reset_count > 0:239            logger.info("Reset %d orphaned verses", reset_count)240241        # Re-queue failed verses too — resuming is the retry mechanism242        # (e.g. after fixing a timeout or a transient API problem)243        failed_reset = 0244        for book in job.books:245            failed_reset += self.state_store.reset_failed_verses(job.job_id, book)246        if failed_reset > 0:247            logger.info("Re-queued %d previously failed verses", failed_reset)248249        return self.run(job.job_id)250251    def get_status(self, job_id: str) -> dict[str, Any]:252        """Get detailed status for a job.253254        Args:255            job_id: Job identifier.256257        Returns:258            Dictionary with job status and progress.259        """260        job = self.state_store.get_bible_job(job_id)261        if not job:262            raise ValueError(f"Job not found: {job_id}")263264        book_jobs = self.state_store.get_book_jobs(job_id)265266        return {267            "job_id": job.job_id,268            "status": job.status.value,269            "created_at": job.created_at.isoformat(),270            "started_at": job.started_at.isoformat() if job.started_at else None,271            "completed_at": job.completed_at.isoformat() if job.completed_at else None,272            "total_books": len(job.books),273            "books": [274                {275                    "book": bj.book,276                    "status": bj.status.value,277                    "total_verses": bj.total_verses,278                    "completed_verses": bj.completed_verses,279                    "failed_verses": bj.failed_verses,280                    "progress": bj.progress_percentage,281                }282                for bj in book_jobs283            ],284            "overall_progress": job.progress_percentage,285        }286287    def _snapshot_config(self) -> dict[str, Any]:288        """Capture current configuration for job snapshot.289290        Returns:291            Dictionary with configuration snapshot.292        """293        config: dict[str, Any] = {}294295        # Load style config if exists296        style_path = self.data_dir / "config" / "style.json"297        if style_path.exists():298            with style_path.open() as f:299                config["style"] = json.load(f)300301        # Add model info302        from src.pipeline.translator import DEFAULT_MODEL303304        config["model"] = DEFAULT_MODEL305306        return config307308    def _initialize_all_books(self, job_id: str, books: list[str]) -> None:309        """Initialize verse jobs for all books.310311        Verses already translated on disk are marked COMPLETED immediately312        so progress bars reflect the true starting position.313314        Args:315            job_id: Job identifier.316            books: List of book names.317        """318        from src.data.loader import DataLoader319        from src.data.storage import Storage320321        loader = DataLoader(self.data_dir)322        storage = Storage(self.data_dir)323        total_verses = 0324325        for book in books:326            # Check if already initialized327            book_job = self.state_store.get_book_job(job_id, book)328            if book_job and book_job.total_verses > 0:329                total_verses += book_job.total_verses330                continue331332            # Load progress once per book to check existing translations333            progress = storage.get_progress(book)334            completed_map = progress.get("completed", {})335336            # Get all verses for this book337            pending_verses: list[tuple[int, int]] = []338            completed_verses: list[tuple[int, int]] = []339            chapters = loader.get_book_chapters(book)340341            for chapter in chapters:342                chapter_verses = loader.load_chapter_verses(book, chapter)343                chapter_completed = completed_map.get(str(chapter), [])344                for verse in chapter_verses:345                    ch = verse.reference.chapter346                    vs = verse.reference.verse347                    if vs in chapter_completed:348                        completed_verses.append((ch, vs))349                    else:350                        pending_verses.append((ch, vs))351352            all_verses = pending_verses + completed_verses353            if all_verses:354                self.state_store.initialize_verse_jobs(355                    job_id, book, pending_verses, completed_verses,356                )357                total_verses += len(all_verses)358                logger.info(359                    "Initialized %s: %d verses (%d already translated)",360                    book, len(all_verses), len(completed_verses),361                )362            else:363                logger.warning("No verses found for %s", book)364365        # Update job total366        self.state_store.update_job_verse_counts(job_id, total=total_verses)367368    def _run_parallel(369        self,370        job_id: str,371        books: list[str],372        rate_limiter: SharedRateLimiter,373        shutdown: GracefulShutdown,374    ) -> list[BookResult]:375        """Run book translations in parallel using threads.376377        Uses ThreadPoolExecutor since workers are I/O-bound (API calls).378        Threads share memory so the rate limiter works without pickling.379380        Args:381            job_id: Job identifier.382            books: List of book names.383            rate_limiter: Shared rate limiter.384            shutdown: Shutdown handler.385386        Returns:387            List of BookResult for each book.388        """389        results: list[BookResult] = []390391        with ThreadPoolExecutor(max_workers=self.max_parallel_books) as executor:392            # Submit all books393            futures = {}394            for book in books:395                if shutdown.should_stop:396                    break397398                future = executor.submit(399                    _translate_book_worker,400                    job_id,401                    book,402                    str(self.data_dir),403                    str(self.state_store.db_path),404                    rate_limiter,405                )406                futures[future] = book407408            # Collect results409            for future in as_completed(futures):410                book = futures[future]411                if shutdown.should_stop:412                    results.append(BookResult(413                        book=book,414                        success=False,415                        verses_translated=0,416                        verses_failed=0,417                        duration_seconds=0,418                        error_message="Cancelled due to shutdown",419                    ))420                    continue421422                try:423                    result = future.result()424                    results.append(result)425                except Exception as e:426                    logger.error("Book %s failed: %s", book, e)427                    results.append(BookResult(428                        book=book,429                        success=False,430                        verses_translated=0,431                        verses_failed=0,432                        duration_seconds=0,433                        error_message=str(e),434                    ))435436        return results437438    def _run_sequential(439        self,440        job_id: str,441        books: list[str],442        rate_limiter: SharedRateLimiter,443        shutdown: GracefulShutdown,444    ) -> list[BookResult]:445        """Run book translations sequentially.446447        Args:448            job_id: Job identifier.449            books: List of book names.450            rate_limiter: Shared rate limiter.451            shutdown: Shutdown handler.452453        Returns:454            List of BookResult for each book.455        """456        results: list[BookResult] = []457458        for book in books:459            if shutdown.should_stop:460                results.append(BookResult(461                    book=book,462                    success=False,463                    verses_translated=0,464                    verses_failed=0,465                    duration_seconds=0,466                    error_message="Cancelled due to shutdown",467                ))468                continue469470            result = _translate_book_worker(471                job_id,472                book,473                str(self.data_dir),474                str(self.state_store.db_path),475                rate_limiter,476            )477            results.append(result)478479        return results480481482def _translate_book_worker(  # pragma: no cover483    job_id: str,484    book: str,485    data_dir_str: str,486    db_path_str: str,487    rate_limiter: SharedRateLimiter | DummyRateLimiter,488) -> BookResult:489    """Worker function to translate a single book.490491    This runs in a separate process when using parallel translation.492493    Args:494        job_id: Job identifier.495        book: Book name to translate.496        data_dir_str: Path to data directory (string for pickling).497        db_path_str: Path to state database (string for pickling).498        rate_limiter: Shared rate limiter.499500    Returns:501        BookResult with translation statistics.502    """503    data_dir = Path(data_dir_str)504    state_store = JobStateStore(db_path_str)505    worker_pid = os.getpid()506507    start_time = time.time()508    verses_translated = 0509    verses_failed = 0510511    # Mark book as in progress512    state_store.update_book_status(513        job_id, book, JobStatus.IN_PROGRESS,514        worker_pid=worker_pid,515        started_at=datetime.utcnow(),516    )517518    try:519        # Create translator components520        from src.config.style_config import StyleConfigManager521        from src.context.builder import ContextBuilder522        from src.context.glossary import GlossaryManager523        from src.context.pericope import PericopeManager524        from src.context.translation_memory import TranslationMemoryManager525        from src.data.loader import DataLoader526        from src.data.storage import Storage527        from src.evaluation.evaluator import TranslationEvaluator528        from src.evaluation.storage import EvaluationStorage529        from src.models.verse import VerseReference530        from src.pipeline.prompt import PromptBuilder531        from src.pipeline.translator import Translator532533        loader = DataLoader(data_dir)534        storage = Storage(data_dir)535536        glossary_terms = loader.load_glossary()537        glossary = GlossaryManager(glossary_terms)538539        style_manager = StyleConfigManager(data_dir / "config" / "style.json")540        style = style_manager.get_style()541542        tm = TranslationMemoryManager(data_dir / "memory" / "segments.json")543        pericope_manager = PericopeManager(data_dir / "metadata" / "pericopes.json")544545        context_builder = ContextBuilder(loader, glossary, pericope_manager=pericope_manager)546        prompt_builder = PromptBuilder(glossary, style=style, translation_memory=tm)547548        evaluator = TranslationEvaluator(data_dir=data_dir)549        eval_storage = EvaluationStorage(data_dir)550551        # Per-verse LLM evaluation doubles API cost/time on whole-Bible runs;552        # opt in with TRANSLATE_BIBLE_EVAL=1 (sampled evaluation after the run553        # is usually the better trade).554        auto_evaluate = os.environ.get("TRANSLATE_BIBLE_EVAL", "0") == "1"555556        translator = Translator(557            context_builder,558            prompt_builder,559            storage,560            translation_memory=tm,561            auto_evaluate=auto_evaluate,562            evaluator=evaluator,563            evaluation_storage=eval_storage,564        )565566        # Get pending verses567        pending = state_store.get_pending_verses(job_id, book)568569        # Pericope/chunk mode is the quality default (see IMPROVEMENT_LOG.md):570        # verse-isolated translation fragments sentences, breaks formulaic571        # series, and drifts referents. Translate every pericope that still572        # has pending verses as one unit; verses not covered by pericope data573        # fall through to the per-verse loop below.574        book_pericopes = pericope_manager.get_book_pericopes(book)575        if book_pericopes:576            pending_set = {(vj.chapter, vj.verse) for vj in pending}577            covered: set[tuple[int, int]] = set()578            for pericope in book_pericopes:579                refs = [tuple(map(int, v.split(":"))) for v in pericope.verses]580                covered.update(refs)581                todo = [rv for rv in refs if rv in pending_set]582                if not todo:583                    continue584                for ch, vs in todo:585                    state_store.atomic_claim_verse(job_id, book, ch, vs, worker_pid)586                try:587                    with rate_limiter:588                        translator.translate_chunk(book, pericope, skip_existing=True)589                    for ch, vs in todo:590                        ref = VerseReference(book=book, chapter=ch, verse=vs)591                        if storage.is_verse_translated(ref):592                            state_store.update_verse_status(593                                job_id, book, ch, vs, JobStatus.COMPLETED594                            )595                            verses_translated += 1596                        else:597                            state_store.update_verse_status(598                                job_id, book, ch, vs, JobStatus.FAILED,599                                error="verse missing after chunk translation",600                            )601                            verses_failed += 1602                except Exception as e:603                    logger.error("Failed pericope %s: %s", pericope.id, e)604                    for ch, vs in todo:605                        state_store.update_verse_status(606                            job_id, book, ch, vs, JobStatus.FAILED, error=str(e)607                        )608                        verses_failed += 1609            pending = [610                vj for vj in pending if (vj.chapter, vj.verse) not in covered611            ]612613        for verse_job in pending:614            # Try to claim the verse atomically615            claimed = state_store.atomic_claim_verse(616                job_id, book, verse_job.chapter, verse_job.verse, worker_pid617            )618            if not claimed:619                continue  # Another worker got it620621            ref = VerseReference(622                book=book, chapter=verse_job.chapter, verse=verse_job.verse623            )624625            # Skip verses already translated on disk626            if storage.is_verse_translated(ref):627                state_store.update_verse_status(628                    job_id, book, verse_job.chapter, verse_job.verse,629                    JobStatus.COMPLETED,630                )631                verses_translated += 1632                continue633634            try:635                # Rate limit the API call636                with rate_limiter:637                    # Translate with provenance tracking638                    output, provenance = _translate_verse_with_provenance(639                        translator, ref, storage, data_dir640                    )641642                # Save provenance643                storage.save_translation_with_provenance(output, provenance)644645                # Mark verse as completed646                state_store.update_verse_status(647                    job_id, book, verse_job.chapter, verse_job.verse,648                    JobStatus.COMPLETED,649                )650                verses_translated += 1651652            except Exception as e:653                logger.error("Failed to translate %s: %s", ref, e)654                state_store.update_verse_status(655                    job_id, book, verse_job.chapter, verse_job.verse,656                    JobStatus.FAILED, error=str(e),657                )658                verses_failed += 1659660        # Save TM661        tm.save()662663        # Mark book as completed664        duration = time.time() - start_time665        state_store.update_book_status(666            job_id, book, JobStatus.COMPLETED,667            completed_at=datetime.utcnow(),668        )669670        return BookResult(671            book=book,672            success=True,673            verses_translated=verses_translated,674            verses_failed=verses_failed,675            duration_seconds=duration,676        )677678    except Exception as e:679        logger.error("Book %s failed: %s", book, e)680        duration = time.time() - start_time681682        state_store.update_book_status(683            job_id, book, JobStatus.FAILED,684            completed_at=datetime.utcnow(),685            error_message=str(e),686        )687688        return BookResult(689            book=book,690            success=False,691            verses_translated=verses_translated,692            verses_failed=verses_failed,693            duration_seconds=duration,694            error_message=str(e),695        )696697698def _translate_verse_with_provenance(  # pragma: no cover699    translator: Any,700    reference: VerseReference,701    storage: Any,702    data_dir: Path,703) -> tuple[Any, TranslationProvenance]:704    """Translate a verse and capture full provenance.705706    Args:707        translator: Translator instance.708        reference: Verse reference to translate.709        storage: Storage instance.710        data_dir: Data directory path.711712    Returns:713        Tuple of (TranslationOutput, TranslationProvenance).714    """715    from src.models.verse import TranslationOutput716717    # Track timing718    started_at = datetime.utcnow()719720    # Build context721    context = translator.context_builder.build_context(reference)722723    # Get TM matches724    tm_matches = translator._get_tm_matches(context) or []725726    # Build prompts727    prompt = translator.prompt_builder.build_prompt(context, tm_matches=tm_matches)728    system_prompt = translator.prompt_builder.get_system_prompt()729730    # Build provenance731    provenance = TranslationProvenance(732        model=translator.model,733        prompt_version="v1.0",734        system_prompt=system_prompt,735        user_prompt=prompt,736        started_at=started_at,737        attempt_number=1,738    )739740    # Capture source texts741    verse = context.target_verse742    if verse.greek_text:743        provenance.source_texts["greek"] = verse.greek_text744    if verse.hebrew_text:745        provenance.source_texts["hebrew"] = verse.hebrew_text746    if verse.english_kjv:747        provenance.source_texts["kjv"] = verse.english_kjv748    if verse.english_asv:749        provenance.source_texts["asv"] = verse.english_asv750    if verse.japanese_meiji:751        provenance.source_texts["meiji"] = verse.japanese_meiji752    if verse.japanese_taisho:753        provenance.source_texts["taisho"] = verse.japanese_taisho754755    # Compute source hashes756    provenance.compute_source_hashes()757758    # Capture context verses759    provenance.preceding_verses = [760        str(v.reference) for v in context.preceding_verses761    ]762    provenance.following_verses = [763        str(v.reference) for v in context.following_verses764    ]765766    # Capture chapter summary767    if context.chapter_summary:768        provenance.chapter_summary = context.chapter_summary.summary769770    # Capture book metadata771    if context.book_metadata:772        provenance.book_metadata = {773            "name": context.book_metadata.name_english,774            "testament": context.book_metadata.testament,775            "themes": context.book_metadata.themes,776        }777778    # Capture glossary hash779    provenance.glossary_hash = compute_glossary_hash(context.glossary_terms)780781    # Capture style hash782    if translator.prompt_builder._style:783        style_dict = translator.prompt_builder._style.to_dict()784        provenance.style_config_hash = compute_style_hash(style_dict)785786    # Capture TM matches787    if tm_matches:788        provenance.tm_matches_used = [789            {790                "source": m.segment.source_text,791                "target": m.segment.target_text,792                "similarity": m.similarity,793            }794            for m in tm_matches795        ]796797    # Call Claude798    translation_text = translator._call_claude(system_prompt, prompt)799800    # Complete timing801    completed_at = datetime.utcnow()802    provenance.completed_at = completed_at803804    # Create output805    output = TranslationOutput(806        reference=reference,807        text=translation_text,808        model=translator.model,809        sources_used=list(provenance.source_texts.keys()),810        glossary_terms_applied=[t.original for t in context.glossary_terms],811    )812813    # Add to TM814    translator._add_to_tm(context, translation_text)815816    return output, provenance817