R6A-R4-R12 canonical translation identity and evidence-grounded QA adjudication

core/generation_runner.py

@@ -37,7 +37,7 @@ ProgressCallback = Callable[..., None]
 MIN_FINAL_REFERENCES = 50
 LITERATURE_EVIDENCE_TARGET = 75
 CITATION_PLAN_TARGET = 70
-TRANSLATION_CONTRACT_VERSION = 'hierarchical-multilingual-v3'
+TRANSLATION_CONTRACT_VERSION = 'hierarchical-multilingual-v4'
 
 
 def _clean_trailing_references(text, keywords=("reference", "references", "reference list", "参考文献")):
@@ -866,6 +866,27 @@ class GenerationRunner:
         # Test doubles / legacy adapters still get deterministic invalidation.
         return hashlib.sha256(str(source or '').encode('utf-8')).hexdigest()
 
+    def _translation_identity_bundle(self, agent, source: str) -> dict[str, str]:
+        """Return one canonical identity domain for translation reuse.
+
+        The frozen English source remains the manuscript authority, while the
+        Translation Compiler may consume a canonicalized representation (for
+        example qualitative quote literals become QUOTE semantic slots).  Cache
+        identity must therefore persist both hashes and derive the private Fact
+        Ledger only from the exact compiled source sent to the translator.
+        """
+        raw_source = str(source or '').strip()
+        compiled_source, _pending, _reused = self._prepare_qualitative_translation_source(
+            raw_source, getattr(self, '_qualitative_quote_sources', {}), {}
+        )
+        compiled_source = str(compiled_source or '').strip()
+        return {
+            'source_hash': self._text_hash(raw_source),
+            'compiled_source': compiled_source,
+            'compiled_source_hash': self._text_hash(compiled_source),
+            'fact_ledger_hash': self._translation_ledger_hash(agent, compiled_source),
+        }
+
     @staticmethod
     def _assemble_partial_translation(translation_sections, adaptive_structure, *, include_front=False) -> str:
         mapping = translation_sections if isinstance(translation_sections, dict) else {}
@@ -939,8 +960,10 @@ class GenerationRunner:
         source = str(source_text or '').strip()
         if not source or self._translation_worker is None:
             return False
-        source_hash = self._text_hash(source)
-        fact_ledger_hash = self._translation_ledger_hash(agent, source)
+        identity = self._translation_identity_bundle(agent, source)
+        source_hash = identity['source_hash']
+        compiled_source_hash = identity['compiled_source_hash']
+        fact_ledger_hash = identity['fact_ledger_hash']
         existing = dict((translation_sections or {}).get(unit_key) or {})
         if not isinstance(getattr(self, '_translation_progress_queued', None), set):
             self._translation_progress_queued = set()
@@ -953,13 +976,23 @@ class GenerationRunner:
         if (
             existing.get('qa_status') in {'passed', 'passed_with_warning'}
             and existing.get('source_hash') == source_hash
+            and existing.get('compiled_source_hash') == compiled_source_hash
             and existing.get('target_lang') == str(self._translation_worker_context.get('target_lang') or '')
             and existing.get('translation_contract_version') == TRANSLATION_CONTRACT_VERSION
             and existing.get('fact_ledger_hash') == fact_ledger_hash
             and existing.get('translated_text')
         ):
             return False
-        pending_key = (str(unit_key), source_hash, fact_ledger_hash)
+        if existing and existing.get('translated_text'):
+            reasons = []
+            if existing.get('source_hash') != source_hash: reasons.append('english_source_hash')
+            if existing.get('compiled_source_hash') != compiled_source_hash: reasons.append('compiled_source_hash')
+            if existing.get('fact_ledger_hash') != fact_ledger_hash: reasons.append('fact_ledger_hash')
+            if existing.get('translation_contract_version') != TRANSLATION_CONTRACT_VERSION: reasons.append('contract_version')
+            if existing.get('target_lang') != str(self._translation_worker_context.get('target_lang') or ''): reasons.append('target_lang')
+            if existing.get('qa_status') not in {'passed', 'passed_with_warning'}: reasons.append('qa_status')
+            logging.info('Translation reuse miss | unit=%s phase=enqueue reasons=%s', unit_name, ','.join(reasons) or 'unknown')
+        pending_key = (str(unit_key), source_hash, compiled_source_hash, fact_ledger_hash)
         if pending_key in self._translation_worker_pending:
             return False
         self._translation_progress_inflight.add(str(unit_key))
@@ -968,6 +1001,7 @@ class GenerationRunner:
             'unit_name': str(unit_name),
             'source_text': source,
             'source_hash': source_hash,
+            'compiled_source_hash': compiled_source_hash,
             'fact_ledger_hash': fact_ledger_hash,
             'english_commit_text': str(english_commit_text or ''),
         })
@@ -1010,9 +1044,12 @@ class GenerationRunner:
         source = str(source_text or '').strip()
         if not source:
             return translation_sections, translation_glossary
-        source_hash = self._text_hash(source)
+        identity = self._translation_identity_bundle(agent, source)
+        source_hash = identity['source_hash']
+        compiled_source_hash = identity['compiled_source_hash']
+        canonical_translation_source = identity['compiled_source']
         contract_version = TRANSLATION_CONTRACT_VERSION
-        fact_ledger_hash = self._translation_ledger_hash(agent, source)
+        fact_ledger_hash = identity['fact_ledger_hash']
         existing = dict((translation_sections or {}).get(unit_key) or {})
         if not isinstance(getattr(self, '_translation_progress_queued', None), set):
             self._translation_progress_queued = set()
@@ -1023,6 +1060,7 @@ class GenerationRunner:
         if (
             existing.get('qa_status') in {'passed', 'passed_with_warning'}
             and existing.get('source_hash') == source_hash
+            and existing.get('compiled_source_hash') == compiled_source_hash
             and existing.get('target_lang') == target_lang
             and existing.get('translation_contract_version') == contract_version
             and existing.get('fact_ledger_hash') == fact_ledger_hash
@@ -1030,6 +1068,15 @@ class GenerationRunner:
         ):
             self._translation_progress_inflight.discard(str(unit_key))
             return translation_sections, translation_glossary
+        if existing and existing.get('translated_text'):
+            reasons = []
+            if existing.get('source_hash') != source_hash: reasons.append('english_source_hash')
+            if existing.get('compiled_source_hash') != compiled_source_hash: reasons.append('compiled_source_hash')
+            if existing.get('fact_ledger_hash') != fact_ledger_hash: reasons.append('fact_ledger_hash')
+            if existing.get('translation_contract_version') != contract_version: reasons.append('contract_version')
+            if existing.get('target_lang') != target_lang: reasons.append('target_lang')
+            if existing.get('qa_status') not in {'passed', 'passed_with_warning'}: reasons.append('qa_status')
+            logging.info('Translation reuse miss | unit=%s phase=final_or_direct reasons=%s', unit_name, ','.join(reasons) or 'unknown')
         self._translation_progress_inflight.add(str(unit_key))
         partial = self._assemble_partial_translation(translation_sections, adaptive_structure, include_front=include_front)
         self._publish_target_preview(
@@ -1069,6 +1116,9 @@ class GenerationRunner:
         translation_source, pending_quote_tags, reused_quote_tags = self._prepare_qualitative_translation_source(
             source, getattr(self, "_qualitative_quote_sources", {}), quote_ledger
         )
+        translation_source = str(translation_source or '').strip()
+        if translation_source != canonical_translation_source:
+            raise RuntimeError('Canonical translation source changed with Quote Ledger state.')
 
         def publish_translation_stage(stage):
             value = str(stage or '').strip().casefold()
@@ -1164,6 +1214,7 @@ class GenerationRunner:
             error_code = str(code_method(error) if callable(code_method) else 'TRANSLATION_PENDING')
             updated_sections[unit_key] = {
                 'source_hash': source_hash,
+                'compiled_source_hash': compiled_source_hash,
                 'target_lang': target_lang,
                 'glossary_version': self._glossary_version(translation_glossary),
                 'translation_contract_version': contract_version,
@@ -1186,12 +1237,16 @@ class GenerationRunner:
             )
             return updated_sections, translation_glossary
         updated_glossary = dict(result.get('glossary') or translation_glossary or {})
+        result_fact_ledger_hash = str(result.get('fact_ledger_hash') or fact_ledger_hash)
+        if result_fact_ledger_hash != fact_ledger_hash:
+            raise RuntimeError('Translation result Fact Ledger identity differs from canonical compiled source.')
         updated_sections[unit_key] = {
             'source_hash': source_hash,
+            'compiled_source_hash': compiled_source_hash,
             'target_lang': target_lang,
             'glossary_version': self._glossary_version(updated_glossary),
             'translation_contract_version': str(result.get('translation_contract_version') or contract_version),
-            'fact_ledger_hash': str(result.get('fact_ledger_hash') or fact_ledger_hash),
+            'fact_ledger_hash': result_fact_ledger_hash,
             'translated_text': str(result.get('text') or '').strip(),
             'qa_status': str(result.get('qa_status') or 'passed'),
             'language_adapter': str(result.get('language_adapter') or ''),

core/pipeline_agent.py

@@ -61,7 +61,7 @@ class CitationAuditError(DeepSeekRequestError):
 
 
 class SSciPipelineAgent:
-    TRANSLATION_CONTRACT_VERSION = 'hierarchical-multilingual-v3'
+    TRANSLATION_CONTRACT_VERSION = 'hierarchical-multilingual-v4'
 
     def __init__(self, ds_key, s2_key=""):
         self.ds_key = str(ds_key or "").strip()
@@ -5300,6 +5300,154 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
         _order, segments = self._parse_translation_claim_segments(restored, [claim_id])
         return segments[claim_id]
 
+    def _adjudicate_persistent_translation_issues(
+        self, sources, translations, issues, target_lang, glossary=None, section_name=''
+    ):
+        """Independently verify persistent stochastic QA findings before fatality.
+
+        Structural/factual invariants remain owned by deterministic validators.
+        This verifier only decides whether an LLM-raised semantic/linguistic
+        issue that survived bounded claim repair has blocking evidence in the
+        source/target pair.  It cannot rewrite text and returns bounded codes.
+        """
+        issue_map = {
+            str(block_id): {
+                str(claim_id): [str(code or '').strip().upper() for code in (codes or []) if str(code or '').strip()]
+                for claim_id, codes in dict(claim_map or {}).items()
+            }
+            for block_id, claim_map in dict(issues or {}).items()
+        }
+        if not issue_map:
+            return {}
+
+        source_map = {str(key): str(value or '').strip() for key, value in dict(sources or {}).items()}
+        target_map = {str(key): str(value or '').strip() for key, value in dict(translations or {}).items()}
+        profile = self._translation_language_profile(target_lang)
+        rows = []
+        for block_id, claim_map in sorted(issue_map.items()):
+            if block_id not in source_map or block_id not in target_map:
+                raise DeepSeekRequestError(f'Translation QA adjudication block identity mismatch: SP-P{block_id}.')
+            source_claims = {item['claim_id']: item['text'] for item in self._translation_claim_units(source_map[block_id])}
+            claim_order = list(source_claims)
+            _order, target_claims = self._parse_translation_claim_segments(target_map[block_id], claim_order)
+            for claim_id, codes in sorted(claim_map.items()):
+                if claim_id not in source_claims or claim_id not in target_claims:
+                    raise DeepSeekRequestError(f'Translation QA adjudication claim identity mismatch: SP-P{block_id}-{claim_id}.')
+                for code in codes:
+                    # Russian sustained Latin prose already has deterministic
+                    # evidence authority inside audit_translation_blocks.
+                    if code == 'LANGUAGE_CONTAMINATION' and profile.get('language_code') == 'ru':
+                        evidence = self._russian_contamination_evidence(
+                            source_claims[claim_id], target_claims[claim_id], glossary
+                        )
+                        if evidence.get('confirmed'):
+                            rows.append({
+                                'id': f'SP-P{block_id}-{claim_id}', 'block_id': block_id,
+                                'claim_id': claim_id, 'code': code, 'english': source_claims[claim_id],
+                                'target': target_claims[claim_id], 'deterministic_confirmed': True,
+                            })
+                            continue
+                    rows.append({
+                        'id': f'SP-P{block_id}-{claim_id}', 'block_id': block_id,
+                        'claim_id': claim_id, 'code': code, 'english': source_claims[claim_id],
+                        'target': target_claims[claim_id], 'deterministic_confirmed': False,
+                    })
+
+        confirmed = {}
+        model_rows = [row for row in rows if not row.get('deterministic_confirmed')]
+        for row in rows:
+            if row.get('deterministic_confirmed'):
+                confirmed.setdefault(row['block_id'], {}).setdefault(row['claim_id'], []).append(row['code'])
+                self._record_translation_assurance(
+                    'QA_ADJUDICATION_CONFIRMED', target_lang=target_lang, block_id=row['id'],
+                    severity='critical', decision='HOLD_FINAL',
+                    detail={'issue_code': row['code'], 'evidence_code': 'UNTRANSLATED_PROSE'},
+                )
+
+        if not model_rows:
+            return confirmed
+
+        allowed_evidence = {
+            'CLAIM_MISSING','CLAIM_ADDED','POLARITY_CHANGED','CAUSALITY_CHANGED',
+            'UNCERTAINTY_CHANGED','MEANING_CHANGED','CITATION_MOVED','ACRONYM_MISATTACHED',
+            'GRAMMATICAL_VIOLATION','GLOSSARY_CONCEPT_MISMATCH','UNTRANSLATED_PROSE',
+            'ACCEPTABLE_VARIANT','MORPHOLOGICAL_VARIANT','STYLISTIC_PREFERENCE',
+            'QA_DISAGREEMENT','NO_BLOCKING_EVIDENCE',
+        }
+        payload = {
+            'section': str(section_name or 'section'),
+            'target_language': target_lang,
+            'language_adapter': profile.get('adapter_id'),
+            'morphology': profile.get('morphology'),
+            'term_policy': profile.get('term_policy'),
+            'items': [
+                {'id': row['id'], 'issue_code': row['code'], 'english': row['english'], 'target': row['target']}
+                for row in model_rows
+            ],
+        }
+        system_prompt = f"""You are an independent bilingual translation QA adjudicator for {target_lang}.
+This is verification-only: do not rewrite, suggest wording, or emit free-form explanations. Each supplied issue already survived two bounded claim-local repair rounds. Confirm a BLOCKING defect only when the English/target pair itself provides clear evidence. Do not treat stylistic preference or a legitimate grammatical/morphological variant as blocking.
+For Russian, allow required case/number/gender inflection. For Kazakh, allow productive case, possessive, person/plural and vowel-harmony suffixes on the correct term stem. For Chinese, allow natural reordering and sentence reshaping inside the same claim while preserving meaning.
+Return valid JSON only: {{"items":[{{"id":"SP-P000001-C000001","issue_code":"GRAMMAR","decision":"CONFIRMED_BLOCKING","evidence_code":"GRAMMATICAL_VIOLATION"}}]}}.
+Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one of: {sorted(allowed_evidence)}. Return every supplied id+issue_code pair exactly once."""
+        try:
+            response = self.call_deepseek(
+                system_prompt, json.dumps(payload, ensure_ascii=False),
+                max_tokens=max(1200, min(3200, 600 + 180 * len(model_rows))),
+                temp=0.0, model_role='reasoning', thinking=False, allow_fallback=False,
+            )
+            parsed = self._extract_json_from_text(response)
+            returned = parsed.get('items', []) if isinstance(parsed, dict) else []
+            by_key = {}
+            for item in returned:
+                if not isinstance(item, dict):
+                    continue
+                key = (str(item.get('id') or ''), str(item.get('issue_code') or '').strip().upper())
+                by_key[key] = item
+            expected = {(row['id'], row['code']) for row in model_rows}
+            if set(by_key) != expected:
+                raise DeepSeekRequestError(
+                    f'Translation QA adjudication identity mismatch: expected={sorted(expected)}, got={sorted(by_key)}.'
+                )
+            for row in model_rows:
+                item = by_key[(row['id'], row['code'])]
+                decision = str(item.get('decision') or '').strip().upper()
+                evidence_code = str(item.get('evidence_code') or '').strip().upper()
+                if decision not in {'CONFIRMED_BLOCKING', 'NOT_CONFIRMED'} or evidence_code not in allowed_evidence:
+                    raise DeepSeekRequestError(f'Translation QA adjudication returned invalid bounded decision for {row["id"]}.')
+                if decision == 'CONFIRMED_BLOCKING':
+                    confirmed.setdefault(row['block_id'], {}).setdefault(row['claim_id'], []).append(row['code'])
+                    self._record_translation_assurance(
+                        'QA_ADJUDICATION_CONFIRMED', target_lang=target_lang, block_id=row['id'],
+                        severity='critical', decision='HOLD_FINAL',
+                        detail={'issue_code': row['code'], 'evidence_code': evidence_code},
+                    )
+                else:
+                    self._record_translation_assurance(
+                        'QA_ADJUDICATION_NOT_CONFIRMED', target_lang=target_lang, block_id=row['id'],
+                        severity='diagnostic', decision='PASS_WITH_WARNING',
+                        detail={'issue_code': row['code'], 'evidence_code': evidence_code},
+                    )
+        except PipelineCancelledError:
+            raise
+        except Exception as error:
+            # A verifier outage or malformed verifier payload is not evidence
+            # that the translation is wrong. Deterministic integrity gates have
+            # already run and remain authoritative.
+            self._record_translation_assurance(
+                'QA_ADJUDICATION_UNCERTAIN', target_lang=target_lang, block_id=section_name or 'section',
+                severity='diagnostic', decision='PASS_WITH_WARNING',
+                detail={'error_type': type(error).__name__, 'items': len(model_rows)},
+            )
+            logging.warning(
+                'Translation QA adjudication technical fallback: section=%s error=%s',
+                section_name or 'section', self._safe_source_error(error),
+            )
+        return {
+            block_id: {claim_id: list(dict.fromkeys(codes)) for claim_id, codes in claim_map.items() if codes}
+            for block_id, claim_map in confirmed.items() if claim_map
+        }
+
     def _audit_translation_blocks_with_retry(
         self, sources, translations, target_lang, glossary=None, section_name='', max_attempts=2
     ):
@@ -5431,6 +5579,12 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
                         {block_id: translations[block_id] for block_id in sorted(touched_blocks)},
                         target_lang, frozen_glossary, section_name=section_name or 'frozen section', max_attempts=2,
                     )
+                if quality_issues:
+                    emit_stage('qa')
+                    quality_issues = self._adjudicate_persistent_translation_issues(
+                        source_blocks, translations, quality_issues, target_lang, frozen_glossary,
+                        section_name=section_name or 'frozen section',
+                    )
                 if quality_issues:
                     detail = '; '.join(
                         f'SP-P{block_id}-{claim_id}: {",".join(codes)}'
@@ -5438,7 +5592,7 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
                         for claim_id, codes in sorted((claim_map or {}).items())
                     )
                     raise DeepSeekRequestError(
-                        f'Translation claim-local QA failed for {section_name or "section"}: {detail[:1000]}'
+                        f'Translation claim-local QA confirmed blocking for {section_name or "section"}: {detail[:1000]}'
                     )
 
                 # Rebuild after any localized repairs, then apply deterministic
@@ -5475,6 +5629,7 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
                 message = str(error)
                 if (
                     message.startswith('Translation claim-local QA failed')
+                    or message.startswith('Translation claim-local QA confirmed blocking')
                     or message.startswith('Translation introduced unowned factual content')
                     or message.startswith('Translation order-sensitive relation changed')
                     or message.startswith('Translation fact order changed')