"""Main orchestrator for fault-tolerant Bible translation. Coordinates parallel book translation across multiple processes, handling job state, graceful shutdown, and error recovery. """ from __future__ import annotations import json import logging import os import time from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime from pathlib import Path from typing import TYPE_CHECKING, Any from src.orchestration.models import ( ALL_BOOKS, NEW_TESTAMENT_BOOKS, OLD_TESTAMENT_BOOKS, BookResult, JobResult, JobStatus, ) from src.orchestration.provenance import ( TranslationProvenance, compute_glossary_hash, compute_hash, compute_style_hash, ) from src.orchestration.rate_limiter import DummyRateLimiter, SharedRateLimiter from src.orchestration.signals import GracefulShutdown from src.orchestration.state import JobStateStore if TYPE_CHECKING: from src.models.verse import VerseReference logger = logging.getLogger(__name__) class JobOrchestrator: """Orchestrates fault-tolerant Bible translation across multiple processes. Features: - Parallel book translation using multiprocessing - Verse-level progress tracking with SQLite persistence - Graceful shutdown handling (Ctrl+C) - Resume capability from any point - Full provenance tracking Example: orchestrator = JobOrchestrator(data_dir=Path("./data")) # Start new job job_id = orchestrator.create_job(books=["john", "matthew"]) result = orchestrator.run(job_id) # Resume after crash result = orchestrator.resume() """ def __init__( self, data_dir: Path, max_parallel_books: int = 4, max_concurrent_api_calls: int = 5, rate_limit_delay: float = 1.2, ) -> None: """Initialize the job orchestrator. Args: data_dir: Path to the data directory. max_parallel_books: Maximum books to translate in parallel. max_concurrent_api_calls: Maximum concurrent API calls across all workers. rate_limit_delay: Delay between API calls in seconds. """ self.data_dir = Path(data_dir) self.max_parallel_books = max_parallel_books self.max_concurrent_api_calls = max_concurrent_api_calls self.rate_limit_delay = rate_limit_delay # Initialize state store self.state_store = JobStateStore(self.data_dir / "jobs" / "state.db") # Shutdown handler (set during run) self._shutdown: GracefulShutdown | None = None def create_job( self, books: list[str] | None = None, nt_only: bool = False, ot_only: bool = False, skip_completed: bool = True, job_id: str | None = None, ) -> str: """Create a new Bible translation job. Args: books: Specific books to translate (default: all 66). nt_only: Only translate New Testament (27 books). ot_only: Only translate Old Testament (39 books). skip_completed: Skip books that are already fully translated. job_id: Optional specific job ID. Returns: Job ID for the created job. """ # Determine which books to translate if books: selected_books = [b.lower() for b in books] elif nt_only: selected_books = list(NEW_TESTAMENT_BOOKS) elif ot_only: selected_books = list(OLD_TESTAMENT_BOOKS) else: selected_books = list(ALL_BOOKS) # Validate book names invalid = set(selected_books) - set(ALL_BOOKS) if invalid: raise ValueError(f"Invalid book names: {invalid}") # Load current configuration config = self._snapshot_config() # Create job in state store job = self.state_store.create_bible_job( books=selected_books, config=config, job_id=job_id, ) logger.info("Created job %s with %d books", job.job_id, len(selected_books)) return job.job_id def run(self, job_id: str) -> JobResult: """Run a Bible translation job. Args: job_id: Job identifier to run. Returns: JobResult with completion status and statistics. """ job = self.state_store.get_bible_job(job_id) if not job: raise ValueError(f"Job not found: {job_id}") if job.status == JobStatus.COMPLETED: raise ValueError(f"Job already completed: {job_id}") # Mark job as started start_time = datetime.utcnow() self.state_store.update_job_status( job_id, JobStatus.IN_PROGRESS, started_at=start_time ) book_results: list[BookResult] = [] with GracefulShutdown() as shutdown: self._shutdown = shutdown # Create thread-safe rate limiter rate_limiter = SharedRateLimiter( max_concurrent=self.max_concurrent_api_calls, delay=self.rate_limit_delay, ) try: # Initialize verse jobs for all books self._initialize_all_books(job_id, job.books) # Process books in parallel using thread pool if self.max_parallel_books > 1: book_results = self._run_parallel( job_id, job.books, rate_limiter, shutdown ) else: book_results = self._run_sequential( job_id, job.books, rate_limiter, shutdown ) finally: rate_limiter.shutdown() # Compute final statistics end_time = datetime.utcnow() duration = (end_time - start_time).total_seconds() books_completed = sum(1 for r in book_results if r.success) books_failed = len(book_results) - books_completed total_translated = sum(r.verses_translated for r in book_results) total_failed = sum(r.verses_failed for r in book_results) # Determine final status all_success = books_failed == 0 and not shutdown.should_stop final_status = JobStatus.COMPLETED if all_success else JobStatus.FAILED self.state_store.update_job_status( job_id, final_status, completed_at=end_time ) return JobResult( job_id=job_id, success=all_success, books_completed=books_completed, books_failed=books_failed, total_verses_translated=total_translated, total_verses_failed=total_failed, duration_seconds=duration, book_results=book_results, ) def resume(self, job_id: str | None = None) -> JobResult: """Resume an incomplete job. Args: job_id: Optional specific job ID. If None, resumes latest incomplete. Returns: JobResult with completion status. """ if job_id: job = self.state_store.get_bible_job(job_id) if not job: raise ValueError(f"Job not found: {job_id}") else: job = self.state_store.get_latest_incomplete_job() if not job: raise ValueError("No incomplete jobs found") logger.info("Resuming job %s", job.job_id) # Reset any orphaned verses (in_progress but not completed) reset_count = self.state_store.reset_orphaned_verses(job.job_id) if reset_count > 0: logger.info("Reset %d orphaned verses", reset_count) # Re-queue failed verses too — resuming is the retry mechanism # (e.g. after fixing a timeout or a transient API problem) failed_reset = 0 for book in job.books: failed_reset += self.state_store.reset_failed_verses(job.job_id, book) if failed_reset > 0: logger.info("Re-queued %d previously failed verses", failed_reset) return self.run(job.job_id) def get_status(self, job_id: str) -> dict[str, Any]: """Get detailed status for a job. Args: job_id: Job identifier. Returns: Dictionary with job status and progress. """ job = self.state_store.get_bible_job(job_id) if not job: raise ValueError(f"Job not found: {job_id}") book_jobs = self.state_store.get_book_jobs(job_id) return { "job_id": job.job_id, "status": job.status.value, "created_at": job.created_at.isoformat(), "started_at": job.started_at.isoformat() if job.started_at else None, "completed_at": job.completed_at.isoformat() if job.completed_at else None, "total_books": len(job.books), "books": [ { "book": bj.book, "status": bj.status.value, "total_verses": bj.total_verses, "completed_verses": bj.completed_verses, "failed_verses": bj.failed_verses, "progress": bj.progress_percentage, } for bj in book_jobs ], "overall_progress": job.progress_percentage, } def _snapshot_config(self) -> dict[str, Any]: """Capture current configuration for job snapshot. Returns: Dictionary with configuration snapshot. """ config: dict[str, Any] = {} # Load style config if exists style_path = self.data_dir / "config" / "style.json" if style_path.exists(): with style_path.open() as f: config["style"] = json.load(f) # Add model info from src.pipeline.translator import DEFAULT_MODEL config["model"] = DEFAULT_MODEL return config def _initialize_all_books(self, job_id: str, books: list[str]) -> None: """Initialize verse jobs for all books. Verses already translated on disk are marked COMPLETED immediately so progress bars reflect the true starting position. Args: job_id: Job identifier. books: List of book names. """ from src.data.loader import DataLoader from src.data.storage import Storage loader = DataLoader(self.data_dir) storage = Storage(self.data_dir) total_verses = 0 for book in books: # Check if already initialized book_job = self.state_store.get_book_job(job_id, book) if book_job and book_job.total_verses > 0: total_verses += book_job.total_verses continue # Load progress once per book to check existing translations progress = storage.get_progress(book) completed_map = progress.get("completed", {}) # Get all verses for this book pending_verses: list[tuple[int, int]] = [] completed_verses: list[tuple[int, int]] = [] chapters = loader.get_book_chapters(book) for chapter in chapters: chapter_verses = loader.load_chapter_verses(book, chapter) chapter_completed = completed_map.get(str(chapter), []) for verse in chapter_verses: ch = verse.reference.chapter vs = verse.reference.verse if vs in chapter_completed: completed_verses.append((ch, vs)) else: pending_verses.append((ch, vs)) all_verses = pending_verses + completed_verses if all_verses: self.state_store.initialize_verse_jobs( job_id, book, pending_verses, completed_verses, ) total_verses += len(all_verses) logger.info( "Initialized %s: %d verses (%d already translated)", book, len(all_verses), len(completed_verses), ) else: logger.warning("No verses found for %s", book) # Update job total self.state_store.update_job_verse_counts(job_id, total=total_verses) def _run_parallel( self, job_id: str, books: list[str], rate_limiter: SharedRateLimiter, shutdown: GracefulShutdown, ) -> list[BookResult]: """Run book translations in parallel using threads. Uses ThreadPoolExecutor since workers are I/O-bound (API calls). Threads share memory so the rate limiter works without pickling. Args: job_id: Job identifier. books: List of book names. rate_limiter: Shared rate limiter. shutdown: Shutdown handler. Returns: List of BookResult for each book. """ results: list[BookResult] = [] with ThreadPoolExecutor(max_workers=self.max_parallel_books) as executor: # Submit all books futures = {} for book in books: if shutdown.should_stop: break future = executor.submit( _translate_book_worker, job_id, book, str(self.data_dir), str(self.state_store.db_path), rate_limiter, ) futures[future] = book # Collect results for future in as_completed(futures): book = futures[future] if shutdown.should_stop: results.append(BookResult( book=book, success=False, verses_translated=0, verses_failed=0, duration_seconds=0, error_message="Cancelled due to shutdown", )) continue try: result = future.result() results.append(result) except Exception as e: logger.error("Book %s failed: %s", book, e) results.append(BookResult( book=book, success=False, verses_translated=0, verses_failed=0, duration_seconds=0, error_message=str(e), )) return results def _run_sequential( self, job_id: str, books: list[str], rate_limiter: SharedRateLimiter, shutdown: GracefulShutdown, ) -> list[BookResult]: """Run book translations sequentially. Args: job_id: Job identifier. books: List of book names. rate_limiter: Shared rate limiter. shutdown: Shutdown handler. Returns: List of BookResult for each book. """ results: list[BookResult] = [] for book in books: if shutdown.should_stop: results.append(BookResult( book=book, success=False, verses_translated=0, verses_failed=0, duration_seconds=0, error_message="Cancelled due to shutdown", )) continue result = _translate_book_worker( job_id, book, str(self.data_dir), str(self.state_store.db_path), rate_limiter, ) results.append(result) return results def _translate_book_worker( # pragma: no cover job_id: str, book: str, data_dir_str: str, db_path_str: str, rate_limiter: SharedRateLimiter | DummyRateLimiter, ) -> BookResult: """Worker function to translate a single book. This runs in a separate process when using parallel translation. Args: job_id: Job identifier. book: Book name to translate. data_dir_str: Path to data directory (string for pickling). db_path_str: Path to state database (string for pickling). rate_limiter: Shared rate limiter. Returns: BookResult with translation statistics. """ data_dir = Path(data_dir_str) state_store = JobStateStore(db_path_str) worker_pid = os.getpid() start_time = time.time() verses_translated = 0 verses_failed = 0 # Mark book as in progress state_store.update_book_status( job_id, book, JobStatus.IN_PROGRESS, worker_pid=worker_pid, started_at=datetime.utcnow(), ) try: # Create translator components from src.config.style_config import StyleConfigManager from src.context.builder import ContextBuilder from src.context.glossary import GlossaryManager from src.context.pericope import PericopeManager from src.context.translation_memory import TranslationMemoryManager from src.data.loader import DataLoader from src.data.storage import Storage from src.evaluation.evaluator import TranslationEvaluator from src.evaluation.storage import EvaluationStorage from src.models.verse import VerseReference from src.pipeline.prompt import PromptBuilder from src.pipeline.translator import Translator loader = DataLoader(data_dir) storage = Storage(data_dir) glossary_terms = loader.load_glossary() glossary = GlossaryManager(glossary_terms) style_manager = StyleConfigManager(data_dir / "config" / "style.json") style = style_manager.get_style() tm = TranslationMemoryManager(data_dir / "memory" / "segments.json") pericope_manager = PericopeManager(data_dir / "metadata" / "pericopes.json") context_builder = ContextBuilder(loader, glossary, pericope_manager=pericope_manager) prompt_builder = PromptBuilder(glossary, style=style, translation_memory=tm) evaluator = TranslationEvaluator(data_dir=data_dir) eval_storage = EvaluationStorage(data_dir) # Per-verse LLM evaluation doubles API cost/time on whole-Bible runs; # opt in with TRANSLATE_BIBLE_EVAL=1 (sampled evaluation after the run # is usually the better trade). auto_evaluate = os.environ.get("TRANSLATE_BIBLE_EVAL", "0") == "1" translator = Translator( context_builder, prompt_builder, storage, translation_memory=tm, auto_evaluate=auto_evaluate, evaluator=evaluator, evaluation_storage=eval_storage, ) # Get pending verses pending = state_store.get_pending_verses(job_id, book) # Pericope/chunk mode is the quality default (see IMPROVEMENT_LOG.md): # verse-isolated translation fragments sentences, breaks formulaic # series, and drifts referents. Translate every pericope that still # has pending verses as one unit; verses not covered by pericope data # fall through to the per-verse loop below. book_pericopes = pericope_manager.get_book_pericopes(book) if book_pericopes: pending_set = {(vj.chapter, vj.verse) for vj in pending} covered: set[tuple[int, int]] = set() for pericope in book_pericopes: refs = [tuple(map(int, v.split(":"))) for v in pericope.verses] covered.update(refs) todo = [rv for rv in refs if rv in pending_set] if not todo: continue for ch, vs in todo: state_store.atomic_claim_verse(job_id, book, ch, vs, worker_pid) try: with rate_limiter: translator.translate_chunk(book, pericope, skip_existing=True) for ch, vs in todo: ref = VerseReference(book=book, chapter=ch, verse=vs) if storage.is_verse_translated(ref): state_store.update_verse_status( job_id, book, ch, vs, JobStatus.COMPLETED ) verses_translated += 1 else: state_store.update_verse_status( job_id, book, ch, vs, JobStatus.FAILED, error="verse missing after chunk translation", ) verses_failed += 1 except Exception as e: logger.error("Failed pericope %s: %s", pericope.id, e) for ch, vs in todo: state_store.update_verse_status( job_id, book, ch, vs, JobStatus.FAILED, error=str(e) ) verses_failed += 1 pending = [ vj for vj in pending if (vj.chapter, vj.verse) not in covered ] for verse_job in pending: # Try to claim the verse atomically claimed = state_store.atomic_claim_verse( job_id, book, verse_job.chapter, verse_job.verse, worker_pid ) if not claimed: continue # Another worker got it ref = VerseReference( book=book, chapter=verse_job.chapter, verse=verse_job.verse ) # Skip verses already translated on disk if storage.is_verse_translated(ref): state_store.update_verse_status( job_id, book, verse_job.chapter, verse_job.verse, JobStatus.COMPLETED, ) verses_translated += 1 continue try: # Rate limit the API call with rate_limiter: # Translate with provenance tracking output, provenance = _translate_verse_with_provenance( translator, ref, storage, data_dir ) # Save provenance storage.save_translation_with_provenance(output, provenance) # Mark verse as completed state_store.update_verse_status( job_id, book, verse_job.chapter, verse_job.verse, JobStatus.COMPLETED, ) verses_translated += 1 except Exception as e: logger.error("Failed to translate %s: %s", ref, e) state_store.update_verse_status( job_id, book, verse_job.chapter, verse_job.verse, JobStatus.FAILED, error=str(e), ) verses_failed += 1 # Save TM tm.save() # Mark book as completed duration = time.time() - start_time state_store.update_book_status( job_id, book, JobStatus.COMPLETED, completed_at=datetime.utcnow(), ) return BookResult( book=book, success=True, verses_translated=verses_translated, verses_failed=verses_failed, duration_seconds=duration, ) except Exception as e: logger.error("Book %s failed: %s", book, e) duration = time.time() - start_time state_store.update_book_status( job_id, book, JobStatus.FAILED, completed_at=datetime.utcnow(), error_message=str(e), ) return BookResult( book=book, success=False, verses_translated=verses_translated, verses_failed=verses_failed, duration_seconds=duration, error_message=str(e), ) def _translate_verse_with_provenance( # pragma: no cover translator: Any, reference: VerseReference, storage: Any, data_dir: Path, ) -> tuple[Any, TranslationProvenance]: """Translate a verse and capture full provenance. Args: translator: Translator instance. reference: Verse reference to translate. storage: Storage instance. data_dir: Data directory path. Returns: Tuple of (TranslationOutput, TranslationProvenance). """ from src.models.verse import TranslationOutput # Track timing started_at = datetime.utcnow() # Build context context = translator.context_builder.build_context(reference) # Get TM matches tm_matches = translator._get_tm_matches(context) or [] # Build prompts prompt = translator.prompt_builder.build_prompt(context, tm_matches=tm_matches) system_prompt = translator.prompt_builder.get_system_prompt() # Build provenance provenance = TranslationProvenance( model=translator.model, prompt_version="v1.0", system_prompt=system_prompt, user_prompt=prompt, started_at=started_at, attempt_number=1, ) # Capture source texts verse = context.target_verse if verse.greek_text: provenance.source_texts["greek"] = verse.greek_text if verse.hebrew_text: provenance.source_texts["hebrew"] = verse.hebrew_text if verse.english_kjv: provenance.source_texts["kjv"] = verse.english_kjv if verse.english_asv: provenance.source_texts["asv"] = verse.english_asv if verse.japanese_meiji: provenance.source_texts["meiji"] = verse.japanese_meiji if verse.japanese_taisho: provenance.source_texts["taisho"] = verse.japanese_taisho # Compute source hashes provenance.compute_source_hashes() # Capture context verses provenance.preceding_verses = [ str(v.reference) for v in context.preceding_verses ] provenance.following_verses = [ str(v.reference) for v in context.following_verses ] # Capture chapter summary if context.chapter_summary: provenance.chapter_summary = context.chapter_summary.summary # Capture book metadata if context.book_metadata: provenance.book_metadata = { "name": context.book_metadata.name_english, "testament": context.book_metadata.testament, "themes": context.book_metadata.themes, } # Capture glossary hash provenance.glossary_hash = compute_glossary_hash(context.glossary_terms) # Capture style hash if translator.prompt_builder._style: style_dict = translator.prompt_builder._style.to_dict() provenance.style_config_hash = compute_style_hash(style_dict) # Capture TM matches if tm_matches: provenance.tm_matches_used = [ { "source": m.segment.source_text, "target": m.segment.target_text, "similarity": m.similarity, } for m in tm_matches ] # Call Claude translation_text = translator._call_claude(system_prompt, prompt) # Complete timing completed_at = datetime.utcnow() provenance.completed_at = completed_at # Create output output = TranslationOutput( reference=reference, text=translation_text, model=translator.model, sources_used=list(provenance.source_texts.keys()), glossary_terms_applied=[t.original for t in context.glossary_terms], ) # Add to TM translator._add_to_tm(context, translation_text) return output, provenance