R6A-R4-R2 quote translation ledger identity type safety

core/generation_runner.py

@@ -608,6 +608,41 @@ class GenerationRunner:
             and str(item.get("quote") or "").strip()
         }
 
+    @staticmethod
+    def _qualitative_quote_ledger_identity_ok(
+        record, *, target_lang: str, request_hash: str, source_quote_hashes: dict,
+        task_id: str = "", translation_contract_version: str = TRANSLATION_CONTRACT_VERSION,
+    ) -> bool:
+        """Type-safe identity gate for task-level target-language Quote reuse.
+
+        Missing scalar identity fields remain backward-compatible with pre-R4
+        checkpoints. Present fields must be strings and match exactly. Quote
+        hashes must be a mapping and match the current frozen English evidence.
+        execution_epoch is deliberately not part of logical identity so a newer
+        epoch can reuse the same QA-passed target rendering for the same task.
+        """
+        if not isinstance(record, dict) or not isinstance(source_quote_hashes, dict):
+            return False
+
+        def scalar_matches(stored, current: str) -> bool:
+            if stored is None or stored == "":
+                return True
+            return isinstance(stored, str) and stored == str(current or "")
+
+        stored_hashes = record.get("source_quote_hashes")
+        if stored_hashes is None or stored_hashes == "":
+            hashes_match = True
+        else:
+            hashes_match = isinstance(stored_hashes, dict) and stored_hashes == source_quote_hashes
+
+        return (
+            scalar_matches(record.get("task_id"), task_id)
+            and scalar_matches(record.get("target_lang"), target_lang)
+            and scalar_matches(record.get("request_hash"), request_hash)
+            and scalar_matches(record.get("translation_contract_version"), translation_contract_version)
+            and hashes_match
+        )
+
     @classmethod
     def _prepare_qualitative_translation_source(cls, source, quote_sources, quote_ledger):
         """Use one task-level target rendering for every frozen qualitative quote.
@@ -953,20 +988,34 @@ class GenerationRunner:
             english_commit_text=english_commit_text,
         )
         updated_sections = dict(translation_sections or {})
-        quote_ledger_record = dict(updated_sections.get("__quote_ledger__") or {})
+        raw_quote_ledger_record = updated_sections.get("__quote_ledger__")
+        if raw_quote_ledger_record is not None and not isinstance(raw_quote_ledger_record, dict):
+            logging.warning(
+                "Qualitative quote translation ledger ignored malformed record type=%s",
+                type(raw_quote_ledger_record).__name__,
+            )
+        quote_ledger_record = (
+            dict(raw_quote_ledger_record) if isinstance(raw_quote_ledger_record, dict) else {}
+        )
         current_quote_hashes = {
             tag: hashlib.sha256(re.sub(r"\s+", " ", quote).strip().encode("utf-8")).hexdigest()
             for tag, quote in getattr(self, "_qualitative_quote_sources", {}).items()
         }
         current_request_hash = str(getattr(self, "request_hash", "") or "")
-        stored_quote_hashes = quote_ledger_record.get("source_quote_hashes")
-        ledger_identity_ok = (
-            quote_ledger_record.get("target_lang") in {None, "", target_lang}
-            and quote_ledger_record.get("request_hash") in {None, "", current_request_hash}
-            and (stored_quote_hashes in {None, ""} or stored_quote_hashes == current_quote_hashes)
+        current_task_id = str(getattr(getattr(self, "task_dir", None), "name", "") or "")
+        ledger_identity_ok = self._qualitative_quote_ledger_identity_ok(
+            quote_ledger_record,
+            task_id=current_task_id,
+            target_lang=target_lang,
+            request_hash=current_request_hash,
+            translation_contract_version=TRANSLATION_CONTRACT_VERSION,
+            source_quote_hashes=current_quote_hashes,
         )
+        stored_translations = quote_ledger_record.get("translations")
         quote_ledger = (
-            dict(quote_ledger_record.get("translations") or {}) if ledger_identity_ok else {}
+            dict(stored_translations)
+            if ledger_identity_ok and isinstance(stored_translations, dict)
+            else {}
         )
         translation_source, pending_quote_tags, reused_quote_tags = self._prepare_qualitative_translation_source(
             source, getattr(self, "_qualitative_quote_sources", {}), quote_ledger