R3.2.8.6 Hierarchical Multilingual Translation Assurance Compiler

core/generation_runner.py

@@ -21,7 +21,7 @@ from typing import Callable, Any
 from config import APP_CONFIG, AMJ_PROMPTS, ASQ_PROMPTS
 from core.checkpoint_store import TaskCheckpointStore
 from core.meta_extractor import extract_golden_metadata
-from core.pipeline_agent import SSciPipelineAgent
+from core.pipeline_agent import SSciPipelineAgent, PipelineCancelledError
 from core.production_guard import build_validated_package, replace_abstract
 from core.qualitative_engine import get_qualitative_engine
 from core.quantitative_engine import get_quantitative_engine
@@ -33,7 +33,7 @@ ProgressCallback = Callable[..., None]
 MIN_FINAL_REFERENCES = 50
 LITERATURE_EVIDENCE_TARGET = 75
 CITATION_PLAN_TARGET = 70
-TRANSLATION_CONTRACT_VERSION = 'claim-bound-v2'
+TRANSLATION_CONTRACT_VERSION = 'hierarchical-multilingual-v1'
 
 
 def _clean_trailing_references(text, keywords=("reference", "references", "reference list", "参考文献")):
@@ -267,17 +267,18 @@ class GenerationRunner:
         parts = []
         if include_front:
             record = mapping.get('__front_matter__') or {}
-            if record.get('qa_status') == 'passed' and record.get('translated_text'):
+            if record.get('qa_status') in {'passed', 'passed_with_warning'} and record.get('translated_text'):
                 parts.append(str(record['translated_text']).strip())
         for key, _title in GenerationRunner._unique_top_level_sections(adaptive_structure):
             record = mapping.get(key) or {}
-            if record.get('qa_status') == 'passed' and record.get('translated_text'):
+            if record.get('qa_status') in {'passed', 'passed_with_warning'} and record.get('translated_text'):
                 parts.append(str(record['translated_text']).strip())
         return '\n\n'.join(part for part in parts if part).strip()
 
     def _translate_frozen_unit(
         self, agent, *, unit_key: str, unit_name: str, source_text: str, target_lang: str,
         translation_sections: dict, translation_glossary: dict, adaptive_structure, include_front=False,
+        defer_failures=True,
     ):
         source = str(source_text or '').strip()
         if not source:
@@ -287,7 +288,7 @@ class GenerationRunner:
         fact_ledger_hash = self._translation_ledger_hash(agent, source)
         existing = dict((translation_sections or {}).get(unit_key) or {})
         if (
-            existing.get('qa_status') == 'passed'
+            existing.get('qa_status') in {'passed', 'passed_with_warning'}
             and existing.get('source_hash') == source_hash
             and existing.get('target_lang') == target_lang
             and existing.get('translation_contract_version') == contract_version
@@ -297,11 +298,36 @@ class GenerationRunner:
             return translation_sections, translation_glossary
         partial = self._assemble_partial_translation(translation_sections, adaptive_structure, include_front=include_front)
         self._publish_target_preview(partial, f'translating:{unit_name}', force=True, complete=False)
-        result = agent.translate_frozen_section(
-            source, target_lang, glossary=translation_glossary, section_name=unit_name, max_attempts=2
-        )
-        updated_glossary = dict(result.get('glossary') or translation_glossary or {})
         updated_sections = dict(translation_sections or {})
+        try:
+            result = agent.translate_frozen_section(
+                source, target_lang, glossary=translation_glossary, section_name=unit_name, max_attempts=2
+            )
+        except PipelineCancelledError:
+            raise
+        except Exception as error:
+            if not defer_failures:
+                raise
+            code_method = getattr(agent, 'translation_error_code', None)
+            error_code = str(code_method(error) if callable(code_method) else 'TRANSLATION_PENDING')
+            updated_sections[unit_key] = {
+                'source_hash': source_hash,
+                'target_lang': target_lang,
+                'glossary_version': self._glossary_version(translation_glossary),
+                'translation_contract_version': contract_version,
+                'fact_ledger_hash': fact_ledger_hash,
+                'translated_text': '',
+                'qa_status': 'repair_pending',
+                'error_code': error_code,
+                'attempts': int(existing.get('attempts') or 0) + 1,
+                'deferred_at': datetime.now(timezone.utc).isoformat(),
+            }
+            logging.warning(
+                'Translation deferred until final gate | unit=%s code=%s error=%s',
+                unit_name, error_code, type(error).__name__,
+            )
+            return updated_sections, translation_glossary
+        updated_glossary = dict(result.get('glossary') or translation_glossary or {})
         updated_sections[unit_key] = {
             'source_hash': source_hash,
             'target_lang': target_lang,
@@ -309,7 +335,9 @@ class GenerationRunner:
             '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),
             'translated_text': str(result.get('text') or '').strip(),
-            'qa_status': 'passed',
+            'qa_status': str(result.get('qa_status') or 'passed'),
+            'language_adapter': str(result.get('language_adapter') or ''),
+            'diagnostics': list(result.get('diagnostics') or [])[-20:],
             'attempts': int(result.get('attempts') or 1),
             'completed_at': datetime.now(timezone.utc).isoformat(),
         }
@@ -1070,7 +1098,7 @@ Output only final academic prose. Do not report the word count."""
                     agent, unit_key=unit_key, unit_name=unit_name, source_text=source_text,
                     target_lang=target_lang, translation_sections=translation_sections,
                     translation_glossary=translation_glossary, adaptive_structure=adaptive_structure,
-                    include_front=True,
+                    include_front=True, defer_failures=False,
                 )
                 partial = self._assemble_partial_translation(
                     translation_sections, adaptive_structure, include_front=True
@@ -1085,6 +1113,13 @@ Output only final academic prose. Do not report the word count."""
                 pct = min(96, 94 + int(index * 2 / max(1, len(final_units))))
                 self._checkpoint(f'translation_{unit_key}', state, pct)
 
+            unresolved_units = [
+                key for key, _name, _source in final_units
+                if (translation_sections.get(key) or {}).get('qa_status') not in {'passed', 'passed_with_warning'}
+                or not (translation_sections.get(key) or {}).get('translated_text')
+            ]
+            if unresolved_units:
+                raise RuntimeError('translation_final_gate_unresolved:' + ','.join(unresolved_units[:8]))
             full_draft_translated = self._assemble_partial_translation(
                 translation_sections, adaptive_structure, include_front=True
             ).rstrip()

core/pipeline_agent.py

@@ -56,7 +56,7 @@ class CitationAuditError(DeepSeekRequestError):
 
 
 class SSciPipelineAgent:
-    TRANSLATION_CONTRACT_VERSION = 'claim-bound-v2'
+    TRANSLATION_CONTRACT_VERSION = 'hierarchical-multilingual-v1'
 
     def __init__(self, ds_key, s2_key=""):
         self.ds_key = str(ds_key or "").strip()
@@ -120,6 +120,9 @@ class SSciPipelineAgent:
         self.citation_map = {}
         self.used_citations = set()
         self.citation_audit_log = []
+        # R3.2.8.6: translation validators emit hierarchical assurance evidence.
+        # Low-level diagnostics cannot independently terminate a manuscript.
+        self.translation_assurance_log = []
         self.lock = threading.RLock()
         
         self.lit_target_papers = APP_CONFIG["lit_target_papers"]
@@ -273,6 +276,7 @@ class SSciPipelineAgent:
                 "citation_map": self.citation_map, 
                 "used_citations": list(self.used_citations),
                 "citation_audit_log": list(self.citation_audit_log),
+                "translation_assurance_log": list(self.translation_assurance_log),
                 "global_memory": self.global_memory,
                 "s2_access_mode": self._s2_access_mode,
                 "s2_logical_searches": int(self._s2_logical_searches),
@@ -294,6 +298,7 @@ class SSciPipelineAgent:
             self.citation_map = state_dict.get("citation_map", {})
             self.used_citations = set(state_dict.get("used_citations", []))
             self.citation_audit_log = list(state_dict.get("citation_audit_log", []))
+            self.translation_assurance_log = list(state_dict.get("translation_assurance_log", []))
             self.global_memory = state_dict.get("global_memory", self.global_memory)
             loaded_mode = str(state_dict.get("s2_access_mode") or "").strip().lower()
             expected_mode = "key" if self.s2_key else "anonymous"
@@ -3277,12 +3282,104 @@ Output only the repaired section."""
         )['abstract']
 
     @staticmethod
-    def _translation_language_rule(target_lang):
-        if 'Russian' in target_lang:
-            return 'Use idiomatic academic Russian with correct case government, gender-number agreement, aspect, punctuation, and terminology. Reorganize English syntax where necessary and avoid literal English calques.'
-        if 'Kazakh' in target_lang:
-            return 'Use idiomatic academic Kazakh with correct case and possessive suffixes, person/plural endings, vowel harmony, consonant assimilation, natural word order, and correct Kazakh-specific letters. Avoid Russian syntactic calques.'
-        return 'Use formal academic Simplified Chinese with explicit concepts, natural Chinese syntax, clear hierarchy, and logically connected claims. Avoid carrying English nominalization and clause order into Chinese mechanically.'
+    def _translation_language_profile(target_lang):
+        """Language-specific rendering policy over one language-neutral claim/fact IR.
+
+        Surface length and script statistics are diagnostic evidence only.
+        Fact/claim integrity remains language-neutral and authoritative.
+        """
+        name = str(target_lang or '')
+        if 'Russian' in name:
+            return {
+                'adapter_id': 'ru-academic-v1', 'language_code': 'ru',
+                'morphology': 'inflectional', 'script': 'cyrillic',
+                'surface_min_key': 'translation_min_block_ratio_cyrillic',
+                'surface_min_default': 0.65, 'surface_max_default': 2.20,
+                'term_policy': 'Allow grammatically required Russian case, gender, number, and inflection around translated terms while preserving concept identity and protected acronyms.',
+                'contamination_policy': 'Treat sustained non-Russian language contamination as a repair signal; Kazakh-specific letters are diagnostic evidence, not a standalone failure rule.',
+            }
+        if 'Kazakh' in name:
+            return {
+                'adapter_id': 'kk-academic-v1', 'language_code': 'kk',
+                'morphology': 'agglutinative', 'script': 'cyrillic',
+                'surface_min_key': 'translation_min_block_ratio_cyrillic',
+                'surface_min_default': 0.65, 'surface_max_default': 2.20,
+                'term_policy': 'Translate directly from English into academic Kazakh. Allow case, possessive, person, plural and other productive suffixes to attach naturally to translated term stems while preserving concept identity and protected acronyms.',
+                'contamination_policy': 'Treat sustained Russian contamination as a repair signal. Absence of Kazakh-specific letters in a short claim is never sufficient by itself to reject the claim.',
+            }
+        return {
+            'adapter_id': 'zh-academic-v1', 'language_code': 'zh',
+            'morphology': 'analytic', 'script': 'han',
+            'surface_min_key': 'translation_min_block_ratio_zh',
+            'surface_min_default': 0.28, 'surface_max_default': 2.20,
+            'term_policy': 'Use stable academic Chinese terminology while preserving concept identity and protected acronyms. Natural Chinese reordering and sentence merging/splitting are allowed inside the same claim.',
+            'contamination_policy': 'Treat sustained untranslated English prose as a repair signal; normal acronyms, model names and protected terms are allowed.',
+        }
+
+    @classmethod
+    def _translation_language_rule(cls, target_lang):
+        profile = cls._translation_language_profile(target_lang)
+        if profile['language_code'] == 'ru':
+            core = 'Use idiomatic academic Russian with correct case government, gender-number agreement, aspect, punctuation, and terminology. Reorganize English syntax where necessary and avoid literal English calques.'
+        elif profile['language_code'] == 'kk':
+            core = 'Use idiomatic academic Kazakh with correct case and possessive suffixes, person/plural endings, vowel harmony, consonant assimilation, natural word order, and correct Kazakh-specific letters. Translate directly from English and avoid Russian syntactic calques.'
+        else:
+            core = 'Use formal academic Simplified Chinese with explicit concepts, natural Chinese syntax, clear hierarchy, and logically connected claims. Avoid carrying English nominalization and clause order into Chinese mechanically.'
+        return core + ' ' + profile['term_policy'] + ' ' + profile['contamination_policy']
+
+    @classmethod
+    def _compile_translation_ir(cls, source):
+        """Compile a private, language-neutral translation intermediate representation."""
+        protected_source, fact_ledger, claim_sources = cls._protect_translation_claims(source)
+        return {
+            'ir_version': 'translation-ir-v1',
+            'protected_source': protected_source,
+            'fact_ledger': fact_ledger,
+            'claim_sources': claim_sources,
+            'claim_ids': list(claim_sources),
+        }
+
+    def _record_translation_assurance(self, code, *, target_lang='', block_id='', severity='diagnostic', decision='PASS_WITH_WARNING', detail=None):
+        row = {
+            'code': str(code or 'TRANSLATION_DIAGNOSTIC'),
+            'language': self._translation_language_profile(target_lang).get('language_code'),
+            'block_id': str(block_id or ''),
+            'severity': str(severity or 'diagnostic'),
+            'decision': str(decision or 'PASS_WITH_WARNING'),
+        }
+        if isinstance(detail, dict):
+            row['detail'] = {str(k): v for k, v in detail.items() if isinstance(v, (str, int, float, bool))}
+        self.translation_assurance_log.append(row)
+        logging.warning(
+            'Translation assurance | code=%s language=%s block=%s severity=%s decision=%s detail=%s',
+            row['code'], row['language'], row['block_id'], row['severity'], row['decision'], row.get('detail', {}),
+        )
+        return row
+
+    @staticmethod
+    def _translation_assurance_decision(evidence):
+        """Central decision manager: validators provide evidence; this method owns consequence."""
+        items = list(evidence or [])
+        if any(str(item.get('severity')) == 'critical' for item in items):
+            return 'HOLD_FINAL'
+        if any(str(item.get('severity')) == 'repair' for item in items):
+            return 'REPAIR_BLOCK'
+        if items:
+            return 'PASS_WITH_WARNING'
+        return 'PASS'
+
+    @classmethod
+    def translation_error_code(cls, error):
+        message = str(error or '')
+        if 'unowned factual content' in message or 'protected placeholder mismatch' in message or 'fact moved across claim boundary' in message or 'fact order changed' in message:
+            return 'FACT_INTEGRITY'
+        if 'claim marker mismatch' in message or 'marker mismatch' in message or 'heading-tree mismatch' in message:
+            return 'STRUCTURE_INTEGRITY'
+        if 'claim-local QA failed' in message or 'semantic' in message.lower() or 'caus' in message.lower():
+            return 'SEMANTIC_INTEGRITY'
+        if 'target-language signal' in message.lower() or 'returned unchanged' in message.lower():
+            return 'LANGUAGE_QUALITY'
+        return 'TRANSLATION_TECHNICAL'
 
     @staticmethod
     def _translation_prompt(target_lang):
@@ -3955,26 +4052,40 @@ Output only the repaired section."""
             mapping[acronym] = acronym
         return mapping
 
-    @staticmethod
-    def _validate_translation_language_quality(translated, target_lang):
+    def _validate_translation_language_quality(self, translated, target_lang):
+        """Validate strong target-language identity; keep weak script heuristics diagnostic-only."""
         text = str(translated or '')
         compact = re.sub(r'\s+', '', text)
         if len(compact) < 80:
             return True
-        if 'Chinese' in target_lang:
-            if len(re.findall(r'[\u4e00-\u9fff]', text)) < max(12, int(len(compact) * 0.08)):
-                raise DeepSeekRequestError('Chinese translation quality validation failed: insufficient Chinese text.')
+        profile = self._translation_language_profile(target_lang)
+        if profile['language_code'] == 'zh':
+            han = re.findall(r'[\u4e00-\u9fff]', text)
+            if not han:
+                raise DeepSeekRequestError('Translation target-language signal missing: Chinese text contains no Han characters.')
+            if len(han) < max(12, int(len(compact) * 0.08)):
+                self._record_translation_assurance(
+                    'ZH_LANGUAGE_SIGNAL_LOW', target_lang=target_lang, severity='diagnostic',
+                    decision='PASS_WITH_WARNING', detail={'han_chars': len(han), 'compact_chars': len(compact)},
+                )
             return True
+
         cyrillic = re.findall(r'[А-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі]', text)
         if not cyrillic:
-            raise DeepSeekRequestError('Target-language translation contains no Cyrillic text.')
+            raise DeepSeekRequestError('Translation target-language signal missing: target contains no Cyrillic text.')
         kazakh_specific = re.findall(r'[ӘәҒғҚқҢңӨөҰұҮүҺһІі]', text)
-        if 'Kazakh' in target_lang and len(cyrillic) >= 200:
+        if profile['language_code'] == 'kk' and len(cyrillic) >= 200:
             minimum = max(2, int(len(cyrillic) * 0.002))
             if len(kazakh_specific) < minimum:
-                raise DeepSeekRequestError('Kazakh translation quality validation failed: insufficient Kazakh-specific letters.')
-        if 'Russian' in target_lang and cyrillic and len(kazakh_specific) / len(cyrillic) > 0.02:
-            raise DeepSeekRequestError('Russian translation quality validation failed: excessive Kazakh-specific letters.')
+                self._record_translation_assurance(
+                    'KK_SPECIFIC_LETTER_SIGNAL_LOW', target_lang=target_lang, severity='diagnostic',
+                    decision='PASS_WITH_WARNING', detail={'kazakh_specific': len(kazakh_specific), 'cyrillic': len(cyrillic)},
+                )
+        if profile['language_code'] == 'ru' and cyrillic and len(kazakh_specific) / len(cyrillic) > 0.02:
+            self._record_translation_assurance(
+                'RU_KAZAKH_LETTER_SIGNAL_HIGH', target_lang=target_lang, severity='diagnostic',
+                decision='PASS_WITH_WARNING', detail={'kazakh_specific': len(kazakh_specific), 'cyrillic': len(cyrillic)},
+            )
         return True
 
     def validate_translation_semantic_equivalence(self, source, translated, target_lang, section_name=''):
@@ -4055,8 +4166,11 @@ SECTION: {section_name}
                 'allowed_issue_codes': sorted(allowed_codes),
                 'items': [{'id': row['id'], 'english': row['english'], 'target': row['target']} for row in rows],
             }
+            profile = self._translation_language_profile(target_lang)
             system_prompt = f"""You are a bilingual academic translation QA classifier for {target_lang}.
 Review EACH claim independently. Do not rewrite and do not provide free-form explanations. Check semantic equivalence and language quality: omissions/additions, negation, uncertainty, association-versus-causation strength, terminology, grammar, language contamination, citation attachment, and acronym attachment.
+LANGUAGE ADAPTER: {profile['adapter_id']}; morphology={profile['morphology']}. {profile['term_policy']} {profile['contamination_policy']}
+For Russian, do not flag a terminology error merely because a translated term is inflected for grammatical case/number/gender. For Kazakh, do not flag a terminology error merely because productive suffixes attach to the correct term stem. For Kazakh language identity, absence of Kazakh-specific letters in one short claim is not sufficient evidence of Russian contamination.
 Return valid JSON only in this exact form: {{"items":[{{"id":"SP-P000001-C000001","ok":true,"codes":[]}}]}}.
 If ok=false, codes MUST contain only values from allowed_issue_codes. Never return numbered prose, examples, quotes, or corrective wording. Return every supplied id exactly once."""
             response = self.call_deepseek(
@@ -4184,37 +4298,61 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
         return parsed
 
     def _validate_translation_block(self, source, translated, block_id, target_lang):
-        # R3.2.8.4: factual equality is enforced on the protected skeleton /
-        # private Fact Ledger.  This routine now owns language/coverage and the
-        # long-form/acronym adjacency check only; raw numeric Counter equality is
-        # deliberately non-authoritative because it creates multilingual false
-        # failures and can miss value swaps.
-        for _long_form, acronym in self._translation_longform_acronym_pairs(source):
-            if not re.search(rf'[（(]\s*{re.escape(acronym)}\s*[）)]', str(translated or '')):
-                raise DeepSeekRequestError(
-                    f'Translation detached long-form/acronym pair in {block_id}: {acronym}.'
-                )
+        """Hierarchical multilingual block assurance.
 
+        Fact/claim invariants are enforced earlier by the private ledger. Surface
+        length and weak language statistics are diagnostic only; strong wrong-
+        language/identity signals request bounded local repair.
+        """
         source_compact = re.sub(r'\s+', '', str(source or ''))
         target_compact = re.sub(r'\s+', '', str(translated or ''))
+        profile = self._translation_language_profile(target_lang)
+        evidence = []
+
+        if source_compact and source_compact.casefold() == target_compact.casefold():
+            evidence.append({'code': 'UNCHANGED_SOURCE', 'severity': 'repair'})
+
         if len(source_compact) >= 120:
-            is_chinese = 'Chinese' in target_lang
-            minimum = float(APP_CONFIG.get(
-                'translation_min_block_ratio_zh' if is_chinese else 'translation_min_block_ratio_cyrillic',
-                0.28 if is_chinese else 0.65,
-            ))
-            maximum = float(APP_CONFIG.get('translation_max_block_ratio', 2.20))
+            minimum = float(APP_CONFIG.get(profile['surface_min_key'], profile['surface_min_default']))
+            maximum = float(APP_CONFIG.get('translation_max_block_ratio', profile['surface_max_default']))
             ratio = len(target_compact) / max(1, len(source_compact))
-            if ratio < minimum or ratio > maximum:
-                raise DeepSeekRequestError(
-                    f'Translation coverage ratio is implausible in {block_id}: {ratio:.3f}.'
-                )
-            if source_compact.casefold() == target_compact.casefold():
-                raise DeepSeekRequestError(f'Translation block {block_id} was returned unchanged.')
-            if is_chinese and not re.search(r'[\u4e00-\u9fff]', translated):
-                raise DeepSeekRequestError(f'Translation block {block_id} contains no Chinese text.')
-            if not is_chinese and not re.search(r'[А-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі]', translated):
-                raise DeepSeekRequestError(f'Translation block {block_id} contains no Cyrillic target-language text.')
+            if ratio < minimum:
+                evidence.append({'code': 'SURFACE_RATIO_LOW', 'severity': 'diagnostic', 'ratio': round(ratio, 4), 'threshold': minimum})
+            elif ratio > maximum:
+                evidence.append({'code': 'SURFACE_RATIO_HIGH', 'severity': 'diagnostic', 'ratio': round(ratio, 4), 'threshold': maximum})
+
+        if len(target_compact) >= 40:
+            if profile['language_code'] == 'zh':
+                han = len(re.findall(r'[\u4e00-\u9fff]', str(translated or '')))
+                if han == 0:
+                    evidence.append({'code': 'TARGET_SCRIPT_MISSING', 'severity': 'repair'})
+            else:
+                cyrillic = re.findall(r'[А-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі]', str(translated or ''))
+                if not cyrillic:
+                    evidence.append({'code': 'TARGET_SCRIPT_MISSING', 'severity': 'repair'})
+                else:
+                    kazakh_specific = re.findall(r'[ӘәҒғҚқҢңӨөҰұҮүҺһІі]', str(translated or ''))
+                    if profile['language_code'] == 'kk' and len(cyrillic) >= 200:
+                        minimum_kk = max(2, int(len(cyrillic) * 0.002))
+                        if len(kazakh_specific) < minimum_kk:
+                            evidence.append({'code': 'KK_SPECIFIC_LETTER_SIGNAL_LOW', 'severity': 'diagnostic', 'kazakh_specific': len(kazakh_specific), 'cyrillic': len(cyrillic)})
+                    if profile['language_code'] == 'ru' and len(kazakh_specific) / max(1, len(cyrillic)) > 0.02:
+                        evidence.append({'code': 'RU_KAZAKH_LETTER_SIGNAL_HIGH', 'severity': 'diagnostic', 'kazakh_specific': len(kazakh_specific), 'cyrillic': len(cyrillic)})
+
+        decision = self._translation_assurance_decision(evidence)
+        for item in evidence:
+            self._record_translation_assurance(
+                item.get('code'), target_lang=target_lang, block_id=block_id,
+                severity=item.get('severity'),
+                decision='REPAIR_BLOCK' if item.get('severity') == 'repair' else 'PASS_WITH_WARNING',
+                detail={key: value for key, value in item.items() if key not in {'code', 'severity'}},
+            )
+        if decision == 'REPAIR_BLOCK':
+            codes = [item.get('code') for item in evidence if item.get('severity') == 'repair']
+            raise DeepSeekRequestError(
+                f'Translation assurance requires local block repair in {block_id}: {codes[:5]}.'
+            )
+        return True
 
     def _validate_translation_block_protected(self, block_id, source, translated, target_lang, mapping=None):
         """Validate one restored block after Fact Ledger integrity passed."""
@@ -4228,10 +4366,10 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
         protected_maps = {}
         claim_sources_by_block = {}
         for block_id, source in batch:
-            protected_source, token_map, claim_sources = self._protect_translation_claims(source)
-            protected_batch.append((block_id, source, protected_source))
-            protected_maps[block_id] = token_map
-            claim_sources_by_block[block_id] = claim_sources
+            ir = self._compile_translation_ir(source)
+            protected_batch.append((block_id, source, ir['protected_source']))
+            protected_maps[block_id] = ir['fact_ledger']
+            claim_sources_by_block[block_id] = ir['claim_sources']
         marked_source = '\n\n'.join(
             f'[[SP-P{block_id}]]\n{protected_source}'
             for block_id, _, protected_source in protected_batch
@@ -4399,12 +4537,21 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
             except Exception as error:
                 last_error = error
                 if attempt >= attempts:
-                    raise
+                    self._record_translation_assurance(
+                        'SEMANTIC_QA_UNCERTAIN', target_lang=target_lang, block_id=section_name or 'section',
+                        severity='diagnostic', decision='PASS_WITH_WARNING',
+                        detail={'attempts': attempts, 'error_type': type(error).__name__},
+                    )
+                    logging.warning(
+                        'Translation semantic QA technical fallback: section=%s attempts=%s error=%s',
+                        section_name or 'section', attempts, self._safe_source_error(error),
+                    )
+                    return {}
                 logging.warning(
                     'Translation QA-only retry: section=%s attempt=%s/%s error=%s',
                     section_name or 'section', attempt, attempts, self._safe_source_error(error),
                 )
-        raise last_error or DeepSeekRequestError('Translation QA transport failed.')
+        return {}
 
     def translate_frozen_section(self, section_text, target_lang, glossary=None, section_name='', max_attempts=2):
         """Translate one frozen English unit and publish only after all QA passes."""
@@ -4413,6 +4560,8 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
             raise DeepSeekRequestError('Frozen English translation unit is empty.')
         if re.search(r'(?m)^#\s+References\s*$', value):
             raise DeepSeekRequestError('References must not be sent to the translation model.')
+        assurance_start = len(self.translation_assurance_log)
+        profile = self._translation_language_profile(target_lang)
         frozen_glossary = self.extend_translation_glossary(value, target_lang, glossary)
         units = self._translation_body_units(value)
         body_units = [unit for unit in units if unit['kind'] == 'body']
@@ -4510,13 +4659,16 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
                 output = '\n\n'.join(output_units).strip()
                 self.validate_translation_alignment(value, output)
                 self._validate_translation_language_quality(output, target_lang)
+                diagnostics = list(self.translation_assurance_log[assurance_start:])
                 return {
                     'text': output,
                     'glossary': frozen_glossary,
                     'attempts': attempt,
-                    'qa_status': 'passed',
+                    'qa_status': 'passed_with_warning' if diagnostics else 'passed',
                     'translation_contract_version': self.TRANSLATION_CONTRACT_VERSION,
                     'fact_ledger_hash': self.translation_fact_ledger_hash(value),
+                    'language_adapter': profile['adapter_id'],
+                    'diagnostics': diagnostics[-20:],
                 }
             except PipelineCancelledError:
                 raise
@@ -4529,7 +4681,7 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
                     or message.startswith('Translation alignment')
                     or message.startswith('Translation changed ')
                     or message.startswith('Translation protected ')
-                    or message.startswith('Translation coverage ratio')
+                    or message.startswith('Translation assurance requires local block repair')
                     or message.startswith('Translation block ')
                     or message.startswith('Chinese translation quality validation failed')
                     or message.startswith('Kazakh translation quality validation failed')
@@ -4600,54 +4752,43 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
         self.validate_translation_alignment(value, output)
         return output.strip()
 
-    @staticmethod
-    def validate_translation_alignment(english_text, translated_text):
+    @classmethod
+    def validate_translation_alignment(cls, english_text, translated_text):
         source = str(english_text or '').strip()
         target = str(translated_text or '').strip()
         if not target:
             raise DeepSeekRequestError('Target-language manuscript is empty.')
-        en_tree = [level for level, _ in SSciPipelineAgent._markdown_heading_tree(source)]
-        tr_tree = [level for level, _ in SSciPipelineAgent._markdown_heading_tree(target)]
+        en_tree = [level for level, _ in cls._markdown_heading_tree(source)]
+        tr_tree = [level for level, _ in cls._markdown_heading_tree(target)]
         if en_tree != tr_tree:
             raise DeepSeekRequestError(
                 f'Translation heading-tree mismatch: English={en_tree}, target={tr_tree}.'
             )
-        # Fact integrity is authoritative at the protected-block ledger gate.
-        # Whole-unit raw Counter equality is intentionally not used here: valid
-        # localized dates/number words can change surface digits, while swapped
-        # values can preserve the same Counter.
+        # R3.2.8.6: whole-unit character ratios and Cyrillic subtype signals are
+        # diagnostics only. Claim markers + private Fact Ledger + semantic claim
+        # QA are authoritative for completeness and fidelity.
         source_body = re.split(r'(?m)^#\s+References\s*$', source, maxsplit=1)[0]
         target_body = re.split(r'(?m)^#\s+(?:参考文献|Литература|Пайдаланылған әдебиеттер)\s*$', target, maxsplit=1)[0]
         source_chars = len(re.sub(r'\s+', '', source_body))
         target_chars = len(re.sub(r'\s+', '', target_body))
-        if re.search(r'(?m)^#\s+参考文献\s*$', target):
-            minimum_ratio = float(APP_CONFIG.get('translation_min_block_ratio_zh', 0.28))
-        elif re.search(r'(?m)^#\s+(?:Литература|Пайдаланылған әдебиеттер)\s*$', target):
-            minimum_ratio = float(APP_CONFIG.get('translation_min_block_ratio_cyrillic', 0.65))
-        else:
-            minimum_ratio = float(APP_CONFIG.get('translation_min_coverage_ratio', 0.25))
         ratio = target_chars / source_chars if source_chars else 0.0
-        if source_chars and ratio < minimum_ratio:
-            raise DeepSeekRequestError(
-                f'Translation content coverage is too low: {ratio:.3f}.'
-            )
+        if source_chars and (ratio < 0.20 or ratio > 2.80):
+            logging.warning('Translation whole-unit surface diagnostic | ratio=%.3f | nonfatal=true', ratio)
 
-        # Russian and Kazakh both use Cyrillic, so a generic Cyrillic check is
-        # insufficient. Long Kazakh manuscripts must contain a meaningful
-        # number of Kazakh-specific letters; Russian manuscripts must not be
-        # dominated by those letters. This is deterministic and adds no API cost.
         kazakh_specific = re.findall(r'[ӘәҒғҚқҢңӨөҰұҮүҺһІі]', target_body)
         cyrillic = re.findall(r'[А-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі]', target_body)
-        if re.search(r'(?m)^#\s+Пайдаланылған әдебиеттер\s*$', target):
+        if re.search(r'(?m)^#\s+Пайдаланылған әдебиеттер\s*$', target) and len(cyrillic) >= 200:
             minimum_kazakh = max(8, int(len(cyrillic) * 0.003))
             if len(kazakh_specific) < minimum_kazakh:
-                raise DeepSeekRequestError(
-                    'Kazakh language validation failed: insufficient Kazakh-specific letters.'
+                logging.warning(
+                    'Kazakh whole-unit language signal diagnostic | specific=%s cyrillic=%s | nonfatal=true',
+                    len(kazakh_specific), len(cyrillic),
                 )
-        elif re.search(r'(?m)^#\s+Литература\s*$', target):
-            if cyrillic and len(kazakh_specific) / len(cyrillic) > 0.02:
-                raise DeepSeekRequestError(
-                    'Russian language validation failed: excessive Kazakh-specific letters.'
+        elif re.search(r'(?m)^#\s+Литература\s*$', target) and cyrillic:
+            if len(kazakh_specific) / len(cyrillic) > 0.02:
+                logging.warning(
+                    'Russian whole-unit language signal diagnostic | kazakh_specific=%s cyrillic=%s | nonfatal=true',
+                    len(kazakh_specific), len(cyrillic),
                 )
         return True