software/src/orchestration/provenance.py

1"""Provenance tracking for translation audit trails.23Captures comprehensive input information for every translation,4enabling full reproducibility and audit capability.5"""67from __future__ import annotations89import hashlib10from dataclasses import dataclass, field11from datetime import datetime12from typing import Any131415def compute_hash(text: str | None) -> str:16    """Compute SHA256 hash of text.1718    Args:19        text: Text to hash, or None.2021    Returns:22        Hex-encoded SHA256 hash, or empty string if text is None.23    """24    if not text:25        return ""26    return hashlib.sha256(text.encode("utf-8")).hexdigest()272829@dataclass30class TranslationProvenance:31    """Complete provenance record for a single translation.3233    Tracks all inputs that contributed to a translation, enabling:34    - Full audit trails35    - Reproducibility verification36    - Change detection (via content hashes)37    - Debugging and quality investigation3839    Attributes:40        source_texts: Source texts used (keyed by type: greek, hebrew, kjv, etc.).41        source_hashes: SHA256 hashes of source texts for change detection.42        preceding_verses: List of verse refs used as preceding context.43        following_verses: List of verse refs used as following context.44        pericope_id: ID of the pericope this verse belongs to, if any.45        style_config_hash: Hash of the style configuration used.46        glossary_hash: Hash of glossary terms used.47        prompt_version: Version identifier for the prompt template.48        model: Model used for translation.49        system_prompt: Complete system prompt sent to Claude.50        user_prompt: Complete user prompt sent to Claude.51        tm_matches_used: Translation memory matches used.52        attempt_number: Which attempt this was (1 = first try).53        started_at: When translation started.54        completed_at: When translation completed.55        evaluation_score: Quality score from evaluation (if evaluated).56        evaluation_flags: Quality flags from evaluation.57        chapter_summary: Chapter summary used as context.58        book_metadata: Book metadata used as context.59    """6061    # Source inputs with hashes for change detection62    source_texts: dict[str, str] = field(default_factory=dict)63    source_hashes: dict[str, str] = field(default_factory=dict)6465    # Context inputs66    preceding_verses: list[str] = field(default_factory=list)67    following_verses: list[str] = field(default_factory=list)68    pericope_id: str | None = None6970    # Configuration (frozen at translation time)71    style_config_hash: str = ""72    glossary_hash: str = ""73    prompt_version: str = "v1.0"74    model: str = ""7576    # Full prompts (complete audit trail)77    system_prompt: str = ""78    user_prompt: str = ""7980    # Translation memory inputs81    tm_matches_used: list[dict[str, Any]] = field(default_factory=list)8283    # Execution details84    attempt_number: int = 185    started_at: datetime | None = None86    completed_at: datetime | None = None8788    # Quality89    evaluation_score: float | None = None90    evaluation_flags: list[str] = field(default_factory=list)9192    # Additional context93    chapter_summary: str | None = None94    book_metadata: dict[str, Any] = field(default_factory=dict)9596    def compute_source_hashes(self) -> None:97        """Compute hashes for all source texts."""98        self.source_hashes = {99            key: compute_hash(text) for key, text in self.source_texts.items()100        }101102    def to_dict(self) -> dict[str, Any]:103        """Convert provenance to a dictionary for JSON serialization.104105        Returns:106            Dictionary representation of provenance.107        """108        return {109            "source_texts": self.source_texts,110            "source_hashes": self.source_hashes,111            "preceding_verses": self.preceding_verses,112            "following_verses": self.following_verses,113            "pericope_id": self.pericope_id,114            "style_config_hash": self.style_config_hash,115            "glossary_hash": self.glossary_hash,116            "prompt_version": self.prompt_version,117            "model": self.model,118            "system_prompt": self.system_prompt,119            "user_prompt": self.user_prompt,120            "tm_matches_used": self.tm_matches_used,121            "attempt_number": self.attempt_number,122            "started_at": self.started_at.isoformat() if self.started_at else None,123            "completed_at": self.completed_at.isoformat() if self.completed_at else None,124            "evaluation_score": self.evaluation_score,125            "evaluation_flags": self.evaluation_flags,126            "chapter_summary": self.chapter_summary,127            "book_metadata": self.book_metadata,128        }129130    @classmethod131    def from_dict(cls, data: dict[str, Any]) -> TranslationProvenance:132        """Create provenance from a dictionary.133134        Args:135            data: Dictionary representation.136137        Returns:138            TranslationProvenance instance.139        """140        return cls(141            source_texts=data.get("source_texts", {}),142            source_hashes=data.get("source_hashes", {}),143            preceding_verses=data.get("preceding_verses", []),144            following_verses=data.get("following_verses", []),145            pericope_id=data.get("pericope_id"),146            style_config_hash=data.get("style_config_hash", ""),147            glossary_hash=data.get("glossary_hash", ""),148            prompt_version=data.get("prompt_version", "v1.0"),149            model=data.get("model", ""),150            system_prompt=data.get("system_prompt", ""),151            user_prompt=data.get("user_prompt", ""),152            tm_matches_used=data.get("tm_matches_used", []),153            attempt_number=data.get("attempt_number", 1),154            started_at=(155                datetime.fromisoformat(data["started_at"])156                if data.get("started_at")157                else None158            ),159            completed_at=(160                datetime.fromisoformat(data["completed_at"])161                if data.get("completed_at")162                else None163            ),164            evaluation_score=data.get("evaluation_score"),165            evaluation_flags=data.get("evaluation_flags", []),166            chapter_summary=data.get("chapter_summary"),167            book_metadata=data.get("book_metadata", {}),168        )169170    def has_source_change(self, other: TranslationProvenance) -> bool:171        """Check if source texts have changed from another provenance.172173        Args:174            other: Another provenance to compare against.175176        Returns:177            True if any source hashes differ.178        """179        for key, hash_value in self.source_hashes.items():180            if other.source_hashes.get(key) != hash_value:181                return True182        return False183184    def get_inputs_summary(self) -> dict[str, Any]:185        """Get a summary of inputs for logging/display.186187        Returns:188            Dictionary with input summary.189        """190        return {191            "sources": list(self.source_texts.keys()),192            "preceding_context": len(self.preceding_verses),193            "following_context": len(self.following_verses),194            "pericope": self.pericope_id is not None,195            "tm_matches": len(self.tm_matches_used),196            "model": self.model,197            "attempt": self.attempt_number,198        }199200201def compute_glossary_hash(terms: list[Any]) -> str:202    """Compute hash of glossary terms used.203204    Args:205        terms: List of glossary terms (any type with string representation).206207    Returns:208        SHA256 hash of sorted term representations.209    """210    if not terms:211        return ""212    # Sort by string representation for consistency213    term_strs = sorted(str(t) for t in terms)214    combined = "|".join(term_strs)215    return compute_hash(combined)216217218def compute_style_hash(style_config: dict[str, Any]) -> str:219    """Compute hash of style configuration.220221    Args:222        style_config: Style configuration dictionary.223224    Returns:225        SHA256 hash of JSON-serialized config.226    """227    if not style_config:228        return ""229    import json230231    # Sort keys for consistency232    json_str = json.dumps(style_config, sort_keys=True)233    return compute_hash(json_str)234