R6A-R4-R14 four-language lexical purity and canonical construct ledger

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-v4'
+TRANSLATION_CONTRACT_VERSION = 'hierarchical-multilingual-v5'
 
 
 def _clean_trailing_references(text, keywords=("reference", "references", "reference list", "参考文献")):
@@ -858,6 +858,99 @@ class GenerationRunner:
         encoded = json.dumps(dict(glossary or {}), ensure_ascii=False, sort_keys=True).encode('utf-8')
         return hashlib.sha256(encoded).hexdigest()[:16]
 
+    @staticmethod
+    def _build_canonical_concept_ledger(
+        mode: str, data_metadata: dict, *, keywords='', adaptive_structure=None, research_logic_contract=''
+    ) -> dict:
+        """Freeze task-level research concept identity before body translation.
+
+        This ledger is intentionally conservative: quantitative construct names
+        come directly from frozen metadata; qualitative concept/theme labels come
+        from the Evidence Ledger/outline. Lower-case hyphenated concepts used in
+        the frozen research logic are added because the legacy Title-Case glossary
+        discovery cannot see them. No semantic synonym merging is performed.
+        """
+        items = []
+        seen = set()
+
+        def add(value, kind):
+            term = re.sub(r'\s+', ' ', str(value or '')).strip(' \t\r\n,;:.')
+            if not term or len(term) < 3 or len(term) > 100:
+                return
+            words = re.findall(r"[A-Za-zÀ-ÖØ-öø-ÿА-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі][\w’'\-]*", term, flags=re.UNICODE)
+            if not words or len(words) > 10:
+                return
+            key = re.sub(r'[–—-]+', '-', term).casefold()
+            if key in seen:
+                return
+            seen.add(key)
+            items.append({'source': term, 'kind': str(kind or 'concept')})
+
+        meta = dict(data_metadata or {})
+        if str(mode or '') == 'amj':
+            constructs = meta.get('constructs') if isinstance(meta.get('constructs'), dict) else {}
+            for role in ('predictor', 'mediator', 'outcome'):
+                add(constructs.get(role), f'quantitative_{role}')
+        else:
+            for theme in (meta.get('themes') or []):
+                if not isinstance(theme, dict):
+                    continue
+                add(theme.get('title'), 'qualitative_theme')
+                for concept in (theme.get('concepts') or []):
+                    add(concept, 'qualitative_concept')
+            for row in (meta.get('evidence') or []):
+                if isinstance(row, dict):
+                    add(row.get('concept'), 'qualitative_concept')
+
+        for task in (adaptive_structure or []):
+            if not isinstance(task, dict):
+                continue
+            for concept in (task.get('concepts') or []):
+                add(concept, 'outline_concept')
+
+        # User keywords remain planning seeds, not canonical identities. Use
+        # them only as a fallback when frozen empirical/qualitative metadata and
+        # outline concepts provide no concept labels at all.
+        if not items:
+            for seed in re.split(r'[,;\n|]+', str(keywords or '')):
+                seed = re.sub(r'\s+', ' ', seed).strip()
+                if 1 <= len(seed.split()) <= 6:
+                    add(seed, 'keyword_fallback')
+
+        # Capture explicit lower-case hyphenated academic expressions only when
+        # they recur or carry a research-concept cue. This closes the
+        # context-awareness/context-aware blind spot without freezing generic
+        # methodological adjectives such as cross-sectional by default.
+        hyphenated = re.findall(r'(?<![A-Za-z])(?:[a-z][a-z]{2,}(?:-[a-z][a-z]{2,})+)(?![A-Za-z])', str(research_logic_contract or ''))
+        concept_cues = {
+            'awareness','trust','governance','literacy','fidelity','performance','context','capability',
+            'calibration','memory','acceptance','effectiveness','efficiency','alignment','recommendation',
+        }
+        counts = {}
+        for term in hyphenated:
+            counts[term.casefold()] = counts.get(term.casefold(), 0) + 1
+        for term in hyphenated:
+            parts = set(term.casefold().split('-'))
+            if counts.get(term.casefold(), 0) >= 2 or parts.intersection(concept_cues):
+                add(term, 'logic_hyphenated_concept')
+
+        return {'version': 'concept-ledger-v1', 'items': items}
+
+    @staticmethod
+    def _canonical_concept_writer_directive(ledger: dict) -> str:
+        terms = [str(item.get('source') or '').strip() for item in (dict(ledger or {}).get('items') or []) if isinstance(item, dict)]
+        terms = [term for term in terms if term]
+        if not terms:
+            return ''
+        rendered = '; '.join(terms[:40])
+        return (
+            'CANONICAL CONCEPT IDENTITY: The following research labels are frozen task-level concepts: '
+            + rendered
+            + '. When referring to the same construct or concept, preserve its canonical English label; '
+              'do not silently replace it with a near-synonym that changes concept identity. Ordinary grammatical '
+              'articles, possessives, and surrounding syntax remain free. Distinct listed labels are not synonyms unless the evidence explicitly says so.'
+        )
+
     @staticmethod
     def _translation_ledger_hash(agent, source: str) -> str:
         method = getattr(agent, 'translation_fact_ledger_hash', None)
@@ -1571,6 +1664,10 @@ class GenerationRunner:
             citation_plan = agent.build_claim_evidence_plan(
                 adaptive_structure, topic, synthesis_result, mode=mode, target_unique=CITATION_PLAN_TARGET
             )
+            canonical_concept_ledger = self._build_canonical_concept_ledger(
+                mode, data_metadata, keywords=keywords, adaptive_structure=adaptive_structure,
+                research_logic_contract=research_logic_contract,
+            )
             planned_unique = {
                 item.get("tag")
                 for items in citation_plan.values()
@@ -1596,6 +1693,7 @@ class GenerationRunner:
                     "full_draft_translated": "",
                     "translation_sections": {},
                     "translation_glossary": {},
+                    "canonical_concept_ledger": canonical_concept_ledger,
                     "completed_sections": [],
                     "context_tail": "",
                     "accumulated_deviation": 0.0,
@@ -1619,6 +1717,13 @@ class GenerationRunner:
         citation_plan = state.get("citation_plan") if isinstance(state.get("citation_plan"), dict) else {}
         translation_sections = dict(state.get('translation_sections') or {})
         translation_glossary = dict(state.get('translation_glossary') or {})
+        canonical_concept_ledger = state.get('canonical_concept_ledger')
+        if not isinstance(canonical_concept_ledger, dict) or not canonical_concept_ledger.get('items'):
+            canonical_concept_ledger = self._build_canonical_concept_ledger(
+                mode, data_metadata, keywords=keywords, adaptive_structure=adaptive_structure,
+                research_logic_contract=research_logic_contract,
+            )
+            state['canonical_concept_ledger'] = canonical_concept_ledger
         working_abstract = str(state.get("working_abstract") or "").strip()
         working_keywords = list(state.get("working_keywords") or [])
         if working_abstract:
@@ -1631,6 +1736,15 @@ class GenerationRunner:
         # glossary order deterministic and avoids sharing mutable writer state.
         translation_agent = SSciPipelineAgent(ds_key, s2_key)
         translation_agent.set_execution_progress_hook(lambda: self._progress_activity())
+        translation_agent.set_cancel_checker(lambda: self.cancel_checker())
+        seeded_glossary = translation_agent.seed_canonical_concept_glossary(
+            canonical_concept_ledger, target_lang, translation_glossary
+        )
+        if seeded_glossary != translation_glossary:
+            translation_glossary = dict(seeded_glossary)
+            state['translation_glossary'] = translation_glossary
+            state['canonical_concept_ledger'] = canonical_concept_ledger
+            self._checkpoint('canonical_concept_ledger_ready', state, 15)
         self._start_translation_worker(
             translation_agent,
             translation_sections=translation_sections,
@@ -1885,6 +1999,7 @@ Planning length: approximately {int(adjusted_target * min_ratio)}-{int(adjusted_
 {str(sec.get('required_content') or '')}
 {qualitative_rule}
 {anti_repeat}
+{self._canonical_concept_writer_directive(canonical_concept_ledger)}
 <PROVISIONAL_RESEARCH_LOGIC>
 {research_logic_contract}
 </PROVISIONAL_RESEARCH_LOGIC>

core/pipeline_agent.py

@@ -61,7 +61,7 @@ class CitationAuditError(DeepSeekRequestError):
 
 
 class SSciPipelineAgent:
-    TRANSLATION_CONTRACT_VERSION = 'hierarchical-multilingual-v4'
+    TRANSLATION_CONTRACT_VERSION = 'hierarchical-multilingual-v5'
 
     def __init__(self, ds_key, s2_key=""):
         self.ds_key = str(ds_key or "").strip()
@@ -3878,6 +3878,143 @@ Output only the repaired section."""
             'source_technical_tokens': int(len(technical)),
         }
 
+    @classmethod
+    def _translation_lexical_purity_evidence(cls, source, target, target_lang, glossary=None):
+        """Detect isolated ordinary English lexemes left in ZH/RU/KK prose.
+
+        This is deliberately narrower than general language identification. It
+        protects research objects, source-matched proper names, and legitimate
+        Latin glossary targets, then treats any remaining ordinary Latin word as
+        untranslated lexical residue. The same gate is shared by Chinese, Russian
+        and Kazakh. Kazakh-vs-Russian discrimination remains evidence-grounded QA
+        because the languages share Cyrillic script and overlapping vocabulary.
+        """
+        profile = cls._translation_language_profile(target_lang)
+        language_code = profile.get('language_code')
+        if language_code not in {'zh', 'ru', 'kk'}:
+            return {'confirmed': False, 'tokens': [], 'language_code': language_code or ''}
+
+        source_value = str(source or '')
+        text = cls._strip_translation_claim_markers(str(target or ''))
+        # Do not exempt source-matched English quotations here: participant
+        # quotations are target-language content and must not remain ordinary
+        # English merely because the same literal exists in the frozen source.
+        text = cls._translation_protected_pattern().sub(' ', text)
+        text = re.sub(r'\[\[SP-(?:P\d{6}|C\d{6}|T\d{6})\]\]', ' ', text)
+
+        english_function_words = {'a','an','and','as','at','by','for','from','in','of','on','or','the','to','with'}
+        legal_latin_terms = set()
+        for value in dict(glossary or {}).values():
+            term = str(value or '').strip()
+            words = re.findall(r'[A-Za-z][A-Za-z0-9+.#-]*', term)
+            title_like = bool(words) and len(words) <= 5 and all(
+                word.casefold() in english_function_words or word[:1].isupper() for word in words
+            )
+            technical_shape = bool(words) and any(
+                any(ch.isdigit() for ch in word) or word.isupper()
+                or (len(word) > 2 and any(ch.isupper() for ch in word[1:])) for word in words
+            )
+            if term and re.search(r'[A-Za-z]', term) and (title_like or technical_shape):
+                legal_latin_terms.add(term)
+        for term in sorted(legal_latin_terms, key=len, reverse=True):
+            text = re.sub(re.escape(term), ' ', text, flags=re.I)
+
+        source_tokens = list(re.finditer(r'\b[A-Za-z][A-Za-z0-9+.#-]*\b', source_value))
+        technical = set()
+        proper = set()
+        ordinary_sentence_starters = {
+            'the','this','these','those','prior','previous','current','our','we','results','findings',
+            'participants','evidence','analysis','study','research','however','therefore','because',
+        }
+        for match in source_tokens:
+            token = match.group(0)
+            if any(ch.isdigit() for ch in token) or token.isupper() or (len(token) > 2 and any(ch.isupper() for ch in token[1:])):
+                technical.add(token)
+                continue
+            if token[:1].isupper() and token.casefold() not in ordinary_sentence_starters:
+                proper.add(token)
+        for token in sorted(technical | proper, key=len, reverse=True):
+            text = re.sub(rf'(?<![A-Za-z0-9_]){re.escape(token)}(?![A-Za-z0-9_])', ' ', text)
+
+        # Source-matched bounded proper titles/institutions remain legal even if
+        # they contain English function words, e.g. University of California.
+        latin_run_pattern = re.compile(r"(?:[A-Za-z][A-Za-z'’-]*(?:[ \t]+|$)){2,8}")
+        for match in list(latin_run_pattern.finditer(text)):
+            candidate = re.sub(r'\s+', ' ', match.group(0)).strip()
+            if not candidate:
+                continue
+            words = candidate.split()
+            title_like = all(word.casefold() in english_function_words or word[:1].isupper() for word in words)
+            if title_like and candidate.casefold() in re.sub(r'\s+', ' ', source_value).casefold():
+                text = text[:match.start()] + (' ' * (match.end() - match.start())) + text[match.end():]
+
+        allow_words = {'et', 'al', 'vs'}
+        residual = []
+        for match in re.finditer(r"[A-Za-z][A-Za-z'’-]*", text):
+            token = match.group(0).strip("-'’–—")
+            folded = token.casefold()
+            if folded in allow_words or len(token) < 4:
+                continue
+            residual.append(token)
+        unique = list(dict.fromkeys(residual))
+        return {
+            'confirmed': bool(unique),
+            'tokens': unique[:8],
+            'token_count': len(residual),
+            'language_code': language_code,
+        }
+
+    @classmethod
+    def _canonical_glossary_violations(cls, source, target, target_lang, glossary=None, canonical_sources=None):
+        """Return canonical concept terms whose frozen target identity disappeared.
+
+        Chinese uses compact exact identity. Russian/Kazakh use conservative
+        lexical stems so grammatical inflection and productive suffixation remain
+        legal while a switch to a different concept word is still detectable.
+        """
+        source_value = str(source or '')
+        target_value = cls._strip_translation_claim_markers(str(target or ''))
+        profile = cls._translation_language_profile(target_lang)
+        language_code = profile.get('language_code')
+        violations = []
+        canonical = {str(item or '').strip().casefold() for item in (canonical_sources or []) if str(item or '').strip()}
+        for source_term, target_term in dict(glossary or {}).items():
+            source_term = str(source_term or '').strip()
+            target_term = str(target_term or '').strip()
+            if canonical and source_term.casefold() not in canonical:
+                continue
+            if not source_term or not target_term or source_term.casefold() not in source_value.casefold():
+                continue
+            # Pure technical/acronym identities are protected elsewhere.
+            compact = re.sub(r'[^A-Za-z0-9]', '', target_term)
+            if compact and ' ' not in target_term and (
+                compact.isupper() or any(ch.isdigit() for ch in compact)
+                or (len(compact) > 2 and any(ch.isupper() for ch in compact[1:]))
+            ):
+                continue
+            if language_code == 'zh':
+                expected = re.sub(r'\s+', '', target_term)
+                actual = re.sub(r'\s+', '', target_value)
+                if expected and expected not in actual:
+                    violations.append(source_term)
+                continue
+            if language_code in {'ru', 'kk'}:
+                expected_words = re.findall(r'[А-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі]{3,}', target_term)
+                actual_words = [word.casefold() for word in re.findall(r'[А-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі]{3,}', target_value)]
+                if not expected_words:
+                    continue
+                missing = False
+                for word in expected_words:
+                    folded = word.casefold()
+                    stem_len = min(len(folded), 5 if len(folded) >= 5 else len(folded))
+                    stem = folded[:stem_len]
+                    if not any(actual.startswith(stem) for actual in actual_words):
+                        missing = True
+                        break
+                if missing:
+                    violations.append(source_term)
+        return list(dict.fromkeys(violations))[:8]
+
     @classmethod
     def _compile_translation_ir(cls, source):
         """Compile a private, language-neutral translation intermediate representation."""
@@ -4625,6 +4762,55 @@ Output only the repaired section."""
             raise DeepSeekRequestError(f'Translation left unresolved protected placeholders in {block_id}.')
         return text.strip() if keep_claim_markers else cls._strip_translation_claim_markers(text)
 
+    def seed_canonical_concept_glossary(self, concept_ledger, target_lang, existing=None):
+        """Translate/freeze canonical task concepts once before section work."""
+        mapping = dict(existing or {})
+        rows = (dict(concept_ledger or {}).get('items') or []) if isinstance(concept_ledger, dict) else []
+        sources = []
+        for row in rows:
+            if not isinstance(row, dict):
+                continue
+            term = re.sub(r'\s+', ' ', str(row.get('source') or '')).strip()
+            if term and term not in sources:
+                sources.append(term)
+        self._canonical_concept_sources = set(sources)
+        if not sources:
+            return mapping
+
+        missing = [term for term in sources if not str(mapping.get(term) or '').strip()]
+        # Acronyms/technical identifiers preserve literal identity; conceptual
+        # phrases must be translated instead of becoming legal English residue.
+        translate_terms = []
+        for term in missing:
+            compact = re.sub(r'[^A-Za-z0-9]', '', term)
+            technical = bool(compact) and (compact.isupper() or any(ch.isdigit() for ch in compact) or (len(compact) > 2 and any(ch.isupper() for ch in compact[1:]))) and ' ' not in term
+            if technical:
+                mapping[term] = term
+            else:
+                translate_terms.append(term)
+        if translate_terms:
+            rule = self._translation_language_rule(target_lang)
+            response = self.call_deepseek(
+                f"Translate the supplied frozen research concepts into {target_lang}. {rule} Each source is a canonical concept identity, not a stylistic phrase. Give one stable canonical target term for each source; do not merge distinct sources or replace one concept with a near-synonym. Return JSON only as {{\"terms\":[{{\"source\":\"...\",\"target\":\"...\"}}]}} and return every source exactly once.",
+                json.dumps({'terms': translate_terms}, ensure_ascii=False),
+                max_tokens=max(1200, min(5000, 120 + len(translate_terms) * 90)),
+                temp=0.0, model_role='translation', thinking=False, allow_fallback=False,
+            )
+            parsed = self._extract_json_from_text(response)
+            items = parsed.get('terms', []) if isinstance(parsed, dict) else []
+            returned = {
+                re.sub(r'\s+', ' ', str(item.get('source') or '')).strip(): re.sub(r'\s+', ' ', str(item.get('target') or '')).strip()
+                for item in items if isinstance(item, dict)
+            }
+            if set(returned) != set(translate_terms) or any(not returned.get(term) for term in translate_terms):
+                raise DeepSeekRequestError('Canonical concept glossary identity mismatch or omitted term.')
+            for term in translate_terms:
+                target = returned[term]
+                if target.casefold() == term.casefold() and re.search(r'[a-z]{4,}', term):
+                    raise DeepSeekRequestError(f'Canonical concept remained untranslated: {term[:80]}')
+                mapping[term] = target
+        return mapping
+
     def extend_translation_glossary(self, body, target_lang, existing=None):
         """Extend a task-level glossary without remapping an existing term."""
         mapping = dict(existing or {})
@@ -4840,7 +5026,7 @@ SECTION: {section_name}
             relevant_glossary = {
                 str(source_term): str(target_term)
                 for source_term, target_term in glossary_value.items()
-                if source_term and str(source_term) in joined_source
+                if source_term and str(source_term).casefold() in joined_source.casefold()
             }
             payload = {
                 'section': str(section_name or 'section'),
@@ -4853,7 +5039,7 @@ SECTION: {section_name}
             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 Russian LANGUAGE_CONTAMINATION, require sustained unprotected non-Russian prose such as a multi-word English clause or sentence. Never use LANGUAGE_CONTAMINATION merely for protected acronyms, AI/LLM/SEM, model or tool names, variable names, citations, glossary terms, source-matched proper names/titles, or isolated Latin terminology. 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.
+For Russian, do not flag a terminology error merely because a translated term is inflected for grammatical case/number/gender. Russian LANGUAGE_CONTAMINATION includes sustained unprotected English prose and ordinary untranslated English lexemes; never use it for protected acronyms, AI/LLM/SEM, model or tool names, variable names, citations, canonical glossary terms, or source-matched proper names/titles. For Chinese, ordinary untranslated English lexical words are contamination while protected technical objects remain legal. For Kazakh, do not flag a terminology error merely because productive suffixes attach to the correct term stem. Kazakh LANGUAGE_CONTAMINATION includes ordinary untranslated English lexemes and clearly Russian lexical/syntactic intrusions not justified by shared international terminology; Cyrillic script or absence of Kazakh-specific letters alone is never sufficient evidence.
 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(
@@ -4877,30 +5063,54 @@ If ok=false, codes MUST contain only values from allowed_issue_codes. Never retu
                 raw_codes = [str(code or '').strip().upper() for code in (returned_row.get('codes') or [])]
                 codes = [code for code in raw_codes if code in allowed_codes]
                 model_contamination = 'LANGUAGE_CONTAMINATION' in codes
+                lexical = self._translation_lexical_purity_evidence(
+                    meta['english'], meta['target'], target_lang, relevant_glossary
+                )
+                canonical_violations = self._canonical_glossary_violations(
+                    meta['english'], meta['target'], target_lang, relevant_glossary,
+                    canonical_sources=getattr(self, '_canonical_concept_sources', set()),
+                )
+                if canonical_violations:
+                    if 'TERMINOLOGY' not in codes:
+                        codes.append('TERMINOLOGY')
+                    self._record_translation_assurance(
+                        'CANONICAL_CONCEPT_MISMATCH', target_lang=target_lang, block_id=row_id,
+                        severity='diagnostic', decision='REPAIR_LOCAL',
+                        detail={'count': len(canonical_violations), 'sources': '|'.join(canonical_violations)[:160]},
+                    )
                 contamination = None
+                deterministic_contamination = bool(lexical.get('confirmed'))
+                deterministic_issue = bool(canonical_violations)
+                if deterministic_contamination:
+                    if 'LANGUAGE_CONTAMINATION' not in codes:
+                        codes.append('LANGUAGE_CONTAMINATION')
+                    self._record_translation_assurance(
+                        'LEXICAL_PURITY_CONFIRMED', target_lang=target_lang, block_id=row_id,
+                        severity='repair', decision='REPAIR_LOCAL',
+                        detail={'token_count': int(lexical.get('token_count') or 0), 'sample': ','.join(lexical.get('tokens') or [])[:120]},
+                    )
                 if language_code == 'ru':
                     contamination = self._russian_contamination_evidence(
                         meta['english'], meta['target'], relevant_glossary
                     )
-                    if contamination.get('confirmed'):
-                        if 'LANGUAGE_CONTAMINATION' not in codes:
-                            codes.append('LANGUAGE_CONTAMINATION')
-                    elif model_contamination:
+                    deterministic_contamination = deterministic_contamination or bool(contamination.get('confirmed'))
+                    if contamination.get('confirmed') and 'LANGUAGE_CONTAMINATION' not in codes:
+                        codes.append('LANGUAGE_CONTAMINATION')
+                    elif model_contamination and not deterministic_contamination:
                         codes = [code for code in codes if code != 'LANGUAGE_CONTAMINATION']
                         self._record_translation_assurance(
                             'RU_LANGUAGE_CONTAMINATION_UNCONFIRMED', target_lang=target_lang, block_id=row_id,
                             severity='diagnostic', decision='PASS_WITH_WARNING', detail=contamination,
                         )
 
-                # A deterministic confirmed Russian contamination signal can
-                # override an LLM false negative. Conversely, a sole Russian
-                # contamination code without deterministic evidence is suppressed.
-                if returned_row.get('ok') is True and not (contamination and contamination.get('confirmed')):
+                # Deterministic lexical residue can override an LLM false negative
+                # in ZH/RU/KK. Russian LLM-only contamination remains suppressed
+                # when neither sustained-prose nor lexical evidence confirms it.
+                if returned_row.get('ok') is True and not deterministic_contamination and not deterministic_issue:
                     continue
                 if not codes:
                     only_suppressed_contamination = (
-                        language_code == 'ru' and model_contamination
-                        and not (contamination or {}).get('confirmed')
+                        language_code == 'ru' and model_contamination and not deterministic_contamination
                         and bool(raw_codes) and all(code == 'LANGUAGE_CONTAMINATION' for code in raw_codes)
                     )
                     if only_suppressed_contamination:
@@ -5334,13 +5544,20 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
                 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
+                    # Lexical purity has deterministic authority in all
+                    # supported target languages. Russian sustained Latin prose
+                    # retains its existing independent deterministic authority.
+                    if code == 'LANGUAGE_CONTAMINATION':
+                        lexical = self._translation_lexical_purity_evidence(
+                            source_claims[claim_id], target_claims[claim_id], target_lang, glossary
                         )
-                        if evidence.get('confirmed'):
+                        deterministic = bool(lexical.get('confirmed'))
+                        if profile.get('language_code') == 'ru' and not deterministic:
+                            evidence = self._russian_contamination_evidence(
+                                source_claims[claim_id], target_claims[claim_id], glossary
+                            )
+                            deterministic = bool(evidence.get('confirmed'))
+                        if deterministic:
                             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],
@@ -5370,7 +5587,7 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
         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',
+            'GRAMMATICAL_VIOLATION','GLOSSARY_CONCEPT_MISMATCH','UNTRANSLATED_PROSE','UNTRANSLATED_LEXEME',
             'ACCEPTABLE_VARIANT','MORPHOLOGICAL_VARIANT','STYLISTIC_PREFERENCE',
             'QA_DISAGREEMENT','NO_BLOCKING_EVIDENCE',
         }
@@ -5387,7 +5604,7 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
         }
         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.
+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, but treat clearly Russian lexical/syntactic substitution as blocking only when the pair itself supports that conclusion. For Chinese, allow natural reordering and sentence reshaping inside the same claim while preserving meaning. For all three languages, an ordinary untranslated English lexical word is blocking; protected acronyms, model/tool names, variables, citations, canonical glossary terms, and source-matched proper names are not.
 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: