S0-R1-C5.5-R6-R6 Canonical Glossary Reviewer Authority Closure

core/pipeline_agent.py

@@ -74,7 +74,7 @@ class SSciPipelineAgent:
     MAX_TRANSLATION_CLAIM_REPAIR_ATTEMPTS = 2
     MAX_TRANSLATION_CLAIM_TOTAL_REPAIRS = 6
     PROTECTED_ENTITY_LEDGER_VERSION = 'protected-entity-ledger-v2'
-    CANONICAL_GLOSSARY_QUALITY_VERSION = 'canonical-glossary-quality-v2'
+    CANONICAL_GLOSSARY_QUALITY_VERSION = 'canonical-glossary-quality-v3'
     SEMANTIC_CLAIM_FRAME_VERSION = 'smartpaper-core-semantic-claim-frame-v1'
     CITATION_CLAIM_FRAME_VERSION = 'smartpaper-citation-claim-frame-v1'
 
@@ -7242,16 +7242,73 @@ TEXT:
             # reduplication is left alone because it can be grammatical Chinese.
             if re.search(r'([\u4e00-\u9fff]{2,4})\1', compact):
                 return True
+        # Canonical labels are short noun phrases. Immediate repetition of the
+        # same substantive token is therefore a high-confidence malformed
+        # surface in Russian/Kazakh (and remains safe for embedded acronyms).
+        words = [
+            token.casefold() for token in re.findall(
+                r"[A-Za-zА-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі][A-Za-zА-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі'’\-]{2,}",
+                value,
+            )
+        ]
+        if any(left == right for left, right in zip(words, words[1:])):
+            return True
         return False
 
-    def validate_and_refine_canonical_concept_glossary(self, concept_ledger, target_lang, glossary):
-        """Independently review strict canonical terms once before body translation.
+    @classmethod
+    def _canonical_target_deterministic_defects(cls, source, target, target_lang):
+        """Return only deterministic defects that make a canonical label unsafe to freeze.
 
-        The first glossary call establishes candidate identities.  This second,
-        bounded call verifies semantic identity and natural academic surface form
-        before those candidates become immutable for all later ZH/RU/KK sections.
-        At most one corrected target is accepted for each strict construct; no
-        section-level repair loop is involved.
+        The model-based quality reviewer is secondary evidence.  It may improve a
+        safe candidate, but it must never turn an otherwise deterministic-safe
+        label into a task-fatal error merely because the review output is
+        contradictory, incomplete, or non-actionable.
+        """
+        source_value = re.sub(r'\s+', ' ', str(source or '')).strip()
+        target_value = re.sub(r'\s+', ' ', str(target or '')).strip()
+        defects = []
+        if not target_value:
+            return ['EMPTY_TARGET']
+
+        compact_source = re.sub(r'[^A-Za-z0-9]', '', source_value)
+        pure_technical_source = bool(compact_source) and ' ' not in source_value and (
+            compact_source.isupper()
+            or any(ch.isdigit() for ch in compact_source)
+            or (len(compact_source) > 2 and any(ch.isupper() for ch in compact_source[1:]))
+        )
+        if not pure_technical_source:
+            if target_value.casefold() == source_value.casefold() and re.search(r'[a-z]{4,}', source_value):
+                defects.append('UNTRANSLATED')
+            language_code = cls._translation_language_profile(target_lang).get('language_code')
+            if language_code == 'zh' and not re.search(r'[\u4e00-\u9fff]', target_value):
+                defects.append('TARGET_LANGUAGE_SIGNAL_MISSING')
+            if language_code in {'ru', 'kk'} and not re.search(r'[А-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі]', target_value):
+                defects.append('TARGET_LANGUAGE_SIGNAL_MISSING')
+
+        if cls._canonical_target_surface_suspicious(target_value, target_lang):
+            defects.append('MALFORMED_SURFACE')
+        if not cls._canonical_target_preserves_source_anchors(source_value, target_value):
+            defects.append('PROTECTED_ANCHOR_LOSS')
+        return list(dict.fromkeys(defects))
+
+    def validate_and_refine_canonical_concept_glossary(self, concept_ledger, target_lang, glossary):
+        """Review strict canonical terms without giving the reviewer unilateral veto power.
+
+        Contract:
+        1. Server-owned deterministic invariants decide whether a candidate is
+           safe enough to freeze (non-empty, translated target-language signal,
+           well-formed surface, protected-anchor preservation).
+        2. The independent model reviewer may replace a candidate only with one
+           distinct, deterministic-safe correction.
+        3. If the reviewer rejects/omits/contradicts a candidate but supplies no
+           usable correction, an already safe candidate is retained with an
+           assurance warning instead of aborting the whole manuscript.
+        4. If the candidate itself is deterministic-unsafe and no safe correction
+           exists, the gate remains fail-closed.
+
+        This prevents a secondary quality-review formatting/judgment failure from
+        becoming a single point of failure across Chinese, Russian, and Kazakh
+        translation while preserving genuine terminology-integrity barriers.
         """
         mapping = dict(glossary or {})
         rows = (dict(concept_ledger or {}).get('items') or []) if isinstance(concept_ledger, dict) else []
@@ -7271,67 +7328,132 @@ TEXT:
             {'source': source, 'candidate_target': str(mapping.get(source) or '').strip()}
             for source in strict_sources
         ]
-        if any(not row['candidate_target'] for row in payload):
-            raise DeepSeekRequestError('Canonical glossary quality review received an empty strict target.')
+        candidate_defects = {
+            row['source']: self._canonical_target_deterministic_defects(
+                row['source'], row['candidate_target'], target_lang
+            )
+            for row in payload
+        }
+        empty_sources = [source for source, defects in candidate_defects.items() if 'EMPTY_TARGET' in defects]
+        if empty_sources:
+            raise DeepSeekRequestError(
+                'Canonical glossary quality review received an empty strict target: ' + ', '.join(empty_sources[:3])
+            )
+
         language_rule = self._translation_language_rule(target_lang)
-        response = self.call_deepseek(
-            (
-                f"Independently review strict research-construct translations for {target_lang}. {language_rule} "
-                "Judge semantic identity and natural academic terminology, not stylistic preference. "
-                "Reject duplicated/malformed wording, concept narrowing/broadening, or a target that changes the construct. "
-                "Every protected exact source anchor inside a construct (for example AI, LLM, model/tool IDs, hypothesis IDs, formulas, statistics or exact numbers) must remain byte-for-byte in the target canonical label; never localize, delete, duplicate, or rename such an anchor. "
-                "If the candidate is sound, return it unchanged with accepted=true. If it is materially flawed, return exactly one corrected canonical target with accepted=false. "
-                "Do not merge distinct constructs. Return JSON only as "
-                '{"terms":[{"source":"...","candidate_target":"...","accepted":true,"target":"..."}]}'
-                " and return every source exactly once."
-            ),
-            json.dumps({'terms': payload}, ensure_ascii=False),
-            max_tokens=max(1200, min(4200, 180 + len(payload) * 120)),
-            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 []
+        try:
+            response = self.call_deepseek(
+                (
+                    f"Independently review strict research-construct translations for {target_lang}. {language_rule} "
+                    "Judge semantic identity and natural academic terminology, not stylistic preference. "
+                    "Reject duplicated/malformed wording, concept narrowing/broadening, or a target that changes the construct. "
+                    "Every protected exact source anchor inside a construct (for example AI, LLM, model/tool IDs, hypothesis IDs, formulas, statistics or exact numbers) must remain byte-for-byte in the target canonical label; never localize, delete, duplicate, or rename such an anchor. "
+                    "If the candidate is sound, return it unchanged with accepted=true. If it is materially flawed, return exactly one corrected canonical target with accepted=false. "
+                    "Do not merge distinct constructs. Return JSON only as "
+                    '{"terms":[{"source":"...","candidate_target":"...","accepted":true,"target":"..."}]}'
+                    " and return every source exactly once."
+                ),
+                json.dumps({'terms': payload}, ensure_ascii=False),
+                max_tokens=max(1200, min(4200, 180 + len(payload) * 120)),
+                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 []
+        except DeepSeekRequestError as exc:
+            unsafe = [source for source, defects in candidate_defects.items() if defects]
+            if unsafe:
+                source = unsafe[0]
+                raise DeepSeekRequestError(
+                    f'Canonical concept has no deterministic-safe target after review failure: {source[:80]} '
+                    f'defects={candidate_defects[source]}'
+                ) from exc
+            self._record_translation_assurance(
+                'CANONICAL_GLOSSARY_REVIEW_UNAVAILABLE', target_lang=target_lang, severity='diagnostic',
+                decision='PASS_WITH_WARNING', detail={'strict_sources': len(strict_sources)},
+            )
+            return mapping
+
         returned = {}
         for item in items:
             if not isinstance(item, dict):
                 continue
             source = re.sub(r'\s+', ' ', str(item.get('source') or '')).strip()
-            if source:
+            if source in strict_sources:
                 returned[source] = item
-        if set(returned) != set(strict_sources):
-            raise DeepSeekRequestError('Canonical glossary quality review omitted or changed a strict source identity.')
+        extra_count = sum(
+            1 for item in items if isinstance(item, dict)
+            and re.sub(r'\s+', ' ', str(item.get('source') or '')).strip() not in strict_sources
+        )
+        if extra_count:
+            self._record_translation_assurance(
+                'CANONICAL_GLOSSARY_REVIEW_EXTRA_SOURCE', target_lang=target_lang, severity='diagnostic',
+                decision='PASS_WITH_WARNING', detail={'extra_sources': extra_count},
+            )
 
         refined = dict(mapping)
         for source in strict_sources:
-            item = returned[source]
             candidate = str(mapping.get(source) or '').strip()
+            defects = list(candidate_defects.get(source) or [])
+            item = returned.get(source)
+            if not item:
+                if defects:
+                    raise DeepSeekRequestError(
+                        f'Canonical concept has no deterministic-safe target after review omission: {source[:80]} '
+                        f'defects={defects}'
+                    )
+                self._record_translation_assurance(
+                    'CANONICAL_GLOSSARY_REVIEW_INCONCLUSIVE', target_lang=target_lang, severity='diagnostic',
+                    decision='PASS_WITH_WARNING', detail={'source': source[:120], 'reason': 'OMITTED_SOURCE'},
+                )
+                continue
+
             accepted_raw = item.get('accepted')
             accepted = accepted_raw is True or str(accepted_raw).strip().casefold() == 'true'
             reviewed_target = re.sub(r'\s+', ' ', str(item.get('target') or '')).strip()
-            suspicious = (
-                self._canonical_target_surface_suspicious(candidate, target_lang)
-                or not self._canonical_target_preserves_source_anchors(source, candidate)
+            reviewed_defects = (
+                self._canonical_target_deterministic_defects(source, reviewed_target, target_lang)
+                if reviewed_target else ['EMPTY_TARGET']
             )
-            if accepted and not suspicious:
-                # An acceptance cannot silently remap the frozen candidate.
-                if reviewed_target and reviewed_target != candidate:
-                    raise DeepSeekRequestError('Canonical glossary review changed an accepted strict target.')
-                continue
-            if not reviewed_target or reviewed_target == candidate:
+
+            # A sound acceptance means no remapping.  If the reviewer claims to
+            # accept but also changes the term, keep a safe frozen candidate; if
+            # the candidate is unsafe, a distinct safe target may still rescue it.
+            if accepted and reviewed_target in {'', candidate}:
+                if not defects:
+                    continue
                 raise DeepSeekRequestError(
-                    f'Canonical concept quality review did not provide a usable correction: {source[:80]}'
+                    f'Canonical concept accepted despite deterministic defects: {source[:80]} defects={defects}'
                 )
-            if reviewed_target.casefold() == source.casefold() and re.search(r'[a-z]{4,}', source):
-                raise DeepSeekRequestError(f'Canonical concept remained untranslated after quality review: {source[:80]}')
-            if (
-                self._canonical_target_surface_suspicious(reviewed_target, target_lang)
-                or not self._canonical_target_preserves_source_anchors(source, reviewed_target)
-            ):
-                raise DeepSeekRequestError(f'Canonical concept correction remains malformed: {source[:80]}')
-            refined[source] = reviewed_target
-            self._record_translation_assurance(
-                'CANONICAL_GLOSSARY_REFINED', target_lang=target_lang, severity='repair',
-                decision='REPAIR_LOCAL', detail={'source': source[:120]},
+
+            distinct_safe_correction = bool(
+                reviewed_target and reviewed_target != candidate and not reviewed_defects
+            )
+            if distinct_safe_correction and (not accepted or defects):
+                refined[source] = reviewed_target
+                self._record_translation_assurance(
+                    'CANONICAL_GLOSSARY_REFINED', target_lang=target_lang, severity='repair',
+                    decision='REPAIR_LOCAL', detail={'source': source[:120]},
+                )
+                continue
+
+            if not defects:
+                # Reviewer vetoes, omits a correction, proposes an invalid
+                # correction, or contradicts accepted=true.  None of these is
+                # actionable evidence against an already deterministic-safe term.
+                reason = 'NON_ACTIONABLE_REJECTION'
+                if accepted and reviewed_target and reviewed_target != candidate:
+                    reason = 'ACCEPTED_BUT_CHANGED'
+                elif reviewed_target and reviewed_target != candidate and reviewed_defects:
+                    reason = 'INVALID_CORRECTION'
+                self._record_translation_assurance(
+                    'CANONICAL_GLOSSARY_REVIEW_INCONCLUSIVE', target_lang=target_lang, severity='diagnostic',
+                    decision='PASS_WITH_WARNING', detail={'source': source[:120], 'reason': reason},
+                )
+                continue
+
+            raise DeepSeekRequestError(
+                f'Canonical concept has no deterministic-safe target after quality review: {source[:80]} '
+                f'defects={defects}'
             )
         return refined