Step 4A-6-C-R3.2.8.1 Translation Protected Recovery Fix

core/pipeline_agent.py

@@ -2397,19 +2397,21 @@ SECTION: {section_name}
         ]
 
     @staticmethod
-    def _numeric_tokens(value):
+    def _translation_numeric_regex():
+        # Canonical immutable numeric token used by both translation protection
+        # and deterministic validation. Covers leading decimals (e.g. .05, -.12),
+        # grouped numbers, percentages, and scientific notation.
+        return r"(?<![A-Za-z])[-+]?(?:(?:\d{1,3}(?:,\d{3})+|\d+)(?:\.\d+)?|\.\d+)(?:[eE][-+]?\d+)?%?"
+
+    @classmethod
+    def _numeric_tokens(cls, value):
         without_citation_tags = re.sub(
             r"\[?CITE-\d+\]?",
             "",
             str(value),
             flags=re.IGNORECASE,
         )
-        return Counter(
-            re.findall(
-                r"(?<![A-Za-z])[-+]?\d+(?:\.\d+)?%?",
-                without_citation_tags,
-            )
-        )
+        return Counter(re.findall(cls._translation_numeric_regex(), without_citation_tags))
 
     @staticmethod
     def _quoted_passages(value):
@@ -3210,10 +3212,17 @@ Output only the revised Discussion body, without the '# Discussion' heading."""
         return Counter(re.findall(r'\bH\d+[A-Za-z]?\b', str(value or ''), flags=re.I))
 
     @staticmethod
-    def _translation_acronym_tokens(value):
-        """Protect exact uppercase abbreviations, including plural suffixes."""
+    def _translation_acronym_regex():
+        # Internal technical hyphens are allowed (GPT-4, PLS-SEM, COVID-19),
+        # but a grammatical hyphen before lowercase prose is not part of the
+        # acronym (AI-powered -> AI; LLM-assisted -> LLM).
+        return r'\b[A-Z]{2,}[A-Z0-9]*(?:-[A-Z0-9]+)*(?:s|es)?\b'
+
+    @classmethod
+    def _translation_acronym_tokens(cls, value):
+        """Return exact immutable uppercase abbreviations, including plurals."""
         without_doi = re.sub(r'https?://\S+|doi:\s*\S+', ' ', str(value or ''), flags=re.I)
-        return Counter(re.findall(r'\b[A-Z][A-Z0-9-]{1,11}(?:s|es)?\b', without_doi))
+        return Counter(re.findall(cls._translation_acronym_regex(), without_doi))
 
     @staticmethod
     def _translation_citation_tokens(value):
@@ -3225,11 +3234,13 @@ Output only the revised Discussion body, without the '# Discussion' heading."""
             )
         )
 
-    @staticmethod
-    def _translation_protected_pattern():
+    @classmethod
+    def _translation_protected_pattern(cls):
         # One-pass alternation prevents a placeholder created for a citation or
-        # URL from being matched again as an acronym/number.  Keep this list
+        # URL from being matched again as an acronym/number. Keep this list
         # conservative: protect immutable research objects, not prose.
+        acronym = cls._translation_acronym_regex()
+        numeric = cls._translation_numeric_regex()
         return re.compile(
             r'(?i:https?://[^\s)>]+)'
             r'|(?i:\bdoi:\s*[^\s)>]+)'
@@ -3241,10 +3252,10 @@ Output only the revised Discussion body, without the '# Discussion' heading."""
             r'|\$[^$\n]+\$'
             r'|(?i:\bH\d+[A-Za-z]?\b)'
             r'|\b(?:fsQCA|SmartPLS|Mplus|OpenAlex|DeepSeek|ChatGPT|RStudio|SPSS|AMOS|Stata)\b'
-            r'|\b[A-Z][A-Z0-9-]{1,11}(?:s|es)?\b'
+            rf'|(?:{acronym})'
             r'|(?:χ²|R²|ΔR²|η²|ω²|[αβγδΔχλμρστφηωΩ])'
             r'|\b(?:QUOTE|CASE|RESP|P)-\d+\b'
-            r'|(?<![A-Za-z])[-+]?\d+(?:\.\d+)?%?'
+            rf'|(?:{numeric})'
             r'|\|'
         )
 
@@ -3288,7 +3299,7 @@ Output only the revised Discussion body, without the '# Discussion' heading."""
         phrases = re.findall(r'\b(?:[A-Z][a-z]+(?:[- ][A-Z][a-z]+){1,3})\b', str(body or ''))
         recurring = [term for term, count in Counter(phrases).most_common() if count >= 2]
         term_limit = max(10, int(APP_CONFIG.get('translation_glossary_term_limit', 40)))
-        sources = list(dict.fromkeys(headings + recurring + acronyms))[:term_limit]
+        sources = list(dict.fromkeys(headings + recurring))[:term_limit]
         missing = [term for term in sources if not str(mapping.get(term) or '').strip()]
         if missing:
             language_rule = self._translation_language_rule(target_lang)
@@ -3310,7 +3321,7 @@ Output only the revised Discussion body, without the '# Discussion' heading."""
             for term in missing:
                 mapping.setdefault(term, new_map[term])
         for acronym in acronyms:
-            mapping.setdefault(acronym, acronym)
+            mapping[acronym] = acronym
         return mapping
 
     @staticmethod
@@ -3477,10 +3488,10 @@ Return valid JSON only: {{"items":[{{"id":"SP-P000001","ok":true,"issues":[]}}]}
         term_limit = max(10, int(APP_CONFIG.get('translation_glossary_term_limit', 40)))
         recurring = [term for term, count in phrase_counts.most_common() if count >= 2]
         unique_headings = list(dict.fromkeys(headings))
-        extras = [term for term in list(dict.fromkeys(recurring + acronyms)) if term not in unique_headings]
+        extras = [term for term in list(dict.fromkeys(recurring)) if term not in unique_headings]
         sources = unique_headings + extras[:max(0, term_limit - len(unique_headings))]
         if not sources:
-            return {}
+            return {acronym: acronym for acronym in acronyms}
 
         language_rule = self._translation_language_rule(target_lang)
         payload = json.dumps({'terms': sources}, ensure_ascii=False)
@@ -3509,13 +3520,16 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
                 'Translation glossary omitted heading(s): ' + ', '.join(missing_headings[:5])
             )
         for acronym in acronyms:
-            mapping.setdefault(acronym, acronym)
+            mapping[acronym] = acronym
         return mapping
 
     @staticmethod
     def _parse_marked_translation(value, expected_ids):
         pattern = re.compile(r'(?m)^\[\[SP-P(\d{6})\]\]\s*$')
-        matches = list(pattern.finditer(str(value or '')))
+        raw_value = str(value or '')
+        matches = list(pattern.finditer(raw_value))
+        if matches and raw_value[:matches[0].start()].strip():
+            raise DeepSeekRequestError('Translation output contained unexpected text before the first SP-P marker.')
         found_ids = [match.group(1) for match in matches]
         if found_ids != list(expected_ids):
             raise DeepSeekRequestError(
@@ -3568,68 +3582,11 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
             if not is_chinese and not re.search(r'[А-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі]', translated):
                 raise DeepSeekRequestError(f'Translation block {block_id} contains no Cyrillic target-language text.')
 
-    def _retry_translation_block_for_abbreviations(self, block_id, source, system_prompt, target_lang):
-        protected_counter = self._translation_acronym_tokens(source)
-        protected = list(protected_counter.elements())
-        strict_prompt = (
-            system_prompt
-            + '\nCRITICAL ABBREVIATION PRESERVATION: Preserve every token in this JSON list '
-              'exactly, including capitalization and plural suffix, and preserve its exact count. '
-              'Do not normalize plural abbreviations (for example, LLMs must not become LLM).\n'
-            + json.dumps(protected, ensure_ascii=False)
-        )
-        marked_source = f'[[SP-P{block_id}]]\n{source}'
-        result = self.call_deepseek(
-            strict_prompt,
-            marked_source,
-            max_tokens=max(1800, min(9000, self._count_tokens(marked_source) * 2 + 1200)),
-            temp=0.0,
-            model_role='translation',
-            thinking=False,
-            allow_fallback=False,
-        )
-        result = self._remove_ai_meta_traces(result)
-        try:
-            translated = self._parse_marked_translation(result, [block_id])[block_id]
-        except DeepSeekRequestError as error:
-            if not str(error).startswith('Translation marker mismatch:'):
-                raise
-            unexpected_markers = re.findall(
-                r'(?m)^\[\[SP-P(\d{6})\]\]\s*$',
-                str(result or ''),
-            )
-            if unexpected_markers:
-                raise DeepSeekRequestError(
-                    'Abbreviation retry returned an unexpected marker: '
-                    f'expected={block_id}, got={unexpected_markers}.'
-                ) from error
-            translated = str(result or '').strip()
-            if not translated:
-                raise DeepSeekRequestError(
-                    f'Abbreviation retry returned empty text for SP-P{block_id}.'
-                ) from error
+    def _validate_translation_block_protected(self, block_id, source, translated, target_lang):
+        """Validate one restored block; recovery is owned by the protected gateway."""
         self._validate_translation_block(source, translated, f'SP-P{block_id}', target_lang)
-        logging.info(
-            'Translation abbreviation recovery accepted block: SP-P%s protected=%s',
-            block_id, protected,
-        )
         return translated
 
-    def _validate_translation_block_with_abbreviation_recovery(self, block_id, source, translated, system_prompt, target_lang):
-        try:
-            self._validate_translation_block(source, translated, f'SP-P{block_id}', target_lang)
-            return translated
-        except DeepSeekRequestError as error:
-            if not str(error).startswith('Translation changed variable abbreviations in SP-P'):
-                raise
-            logging.warning(
-                'Translation abbreviation recovery: block=SP-P%s error=%s',
-                block_id, str(error),
-            )
-            return self._retry_translation_block_for_abbreviations(
-                block_id, source, system_prompt, target_lang
-            )
-
     def _translate_marked_batch_with_recovery(self, batch, system_prompt, target_lang, _strict=False):
         """Translate one marked batch with protected tokens and bounded recovery.
 
@@ -3660,8 +3617,9 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
             max_tokens=max(1800, min(9000, self._count_tokens(marked_source) * 2 + 1200)),
             temp=0.0, model_role='translation', thinking=False, allow_fallback=False,
         )
-        result = self._remove_ai_meta_traces(result)
-
+        # Translation is non-destructive: never run generic manuscript cleanup
+        # on target-language text. Marker/placeholder parsing rejects malformed
+        # output instead of deleting potentially legitimate manuscript lines.
         try:
             parsed = self._parse_marked_translation(result, expected_ids)
         except DeepSeekRequestError as error:
@@ -3701,30 +3659,59 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
             logging.info('Translation marker recovery accepted marker-free single block: SP-P%s', block_id)
 
         restored = {}
-        try:
-            for block_id, source in batch:
+        failed = {}
+        for block_id, source in batch:
+            try:
                 translated = self._restore_translation_tokens(
                     parsed[block_id], protected_maps.get(block_id, {}), f'SP-P{block_id}'
                 )
-                translated = self._validate_translation_block_with_abbreviation_recovery(
-                    block_id, source, translated, system_prompt, target_lang
+                restored[block_id] = self._validate_translation_block_protected(
+                    block_id, source, translated, target_lang
                 )
-                restored[block_id] = translated
-        except DeepSeekRequestError as error:
-            protected_failure = 'protected placeholder mismatch' in str(error) or 'unresolved protected placeholders' in str(error)
-            if protected_failure and len(batch) > 1:
-                middle = max(1, len(batch) // 2)
-                logging.warning('Translation protected-token recovery: batch_size=%s; binary splitting.', len(batch))
-                recovered = {}
-                recovered.update(self._translate_marked_batch_with_recovery(batch[:middle], system_prompt, target_lang))
-                recovered.update(self._translate_marked_batch_with_recovery(batch[middle:], system_prompt, target_lang))
-                return recovered
-            if protected_failure and len(batch) == 1 and not _strict:
-                logging.warning('Translation protected-token recovery: strict single-block retry for SP-P%s.', batch[0][0])
-                return self._translate_marked_batch_with_recovery(batch, system_prompt, target_lang, _strict=True)
-            raise
+            except DeepSeekRequestError as error:
+                failed[block_id] = error
+
+        if failed:
+            if _strict:
+                # A strict isolated retry already used the protected gateway.
+                # Failing again is fail-closed; never fall back to raw source.
+                first_id = next(iter(failed))
+                raise failed[first_id]
+            source_by_id = {block_id: source for block_id, source in batch}
+            for block_id, error in failed.items():
+                logging.warning(
+                    'Translation protected block recovery: block=SP-P%s error=%s; strict isolated retry.',
+                    block_id, self._safe_source_error(error),
+                )
+                recovered = self._translate_marked_batch_with_recovery(
+                    [(block_id, source_by_id[block_id])], system_prompt, target_lang, _strict=True
+                )
+                restored[block_id] = recovered[block_id]
         return restored
 
+    def _audit_translation_blocks_with_retry(
+        self, sources, translations, target_lang, glossary=None, section_name='', max_attempts=2
+    ):
+        """Retry QA transport/JSON failures without retranslating any block."""
+        attempts = max(1, int(max_attempts or 1))
+        last_error = None
+        for attempt in range(1, attempts + 1):
+            try:
+                return self.audit_translation_blocks(
+                    sources, translations, target_lang, glossary, section_name=section_name
+                )
+            except PipelineCancelledError:
+                raise
+            except Exception as error:
+                last_error = error
+                if attempt >= attempts:
+                    raise
+                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.')
+
     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."""
         value = str(section_text or '').strip()
@@ -3765,9 +3752,9 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
                     for batch in batches
                     for block_id, source in batch
                 }
-                quality_issues = self.audit_translation_blocks(
+                quality_issues = self._audit_translation_blocks_with_retry(
                     source_blocks, translations, target_lang, frozen_glossary,
-                    section_name=section_name or 'frozen section',
+                    section_name=section_name or 'frozen section', max_attempts=2,
                 )
                 local_round = 0
                 while quality_issues and local_round < 2:
@@ -3789,10 +3776,10 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
                             [(block_id, source_blocks[block_id])], quality_prompt, target_lang, _strict=True
                         )
                         translations[block_id] = retry[block_id]
-                    quality_issues = self.audit_translation_blocks(
+                    quality_issues = self._audit_translation_blocks_with_retry(
                         {block_id: source_blocks[block_id] for block_id in failed_ids},
                         {block_id: translations[block_id] for block_id in failed_ids},
-                        target_lang, frozen_glossary, section_name=section_name or 'frozen section',
+                        target_lang, frozen_glossary, section_name=section_name or 'frozen section', max_attempts=2,
                     )
                 if quality_issues:
                     detail = '; '.join(
@@ -3829,7 +3816,18 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
                 raise
             except Exception as error:
                 last_error = error
-                if str(error).startswith('Translation block-local QA failed'):
+                message = str(error)
+                if (
+                    message.startswith('Translation block-local QA failed')
+                    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 block ')
+                    or message.startswith('Chinese translation quality validation failed')
+                    or message.startswith('Kazakh translation quality validation failed')
+                    or message.startswith('Russian translation quality validation failed')
+                ):
                     raise
                 logging.warning(
                     'Translation section technical retry: section=%s attempt=%s/%s error=%s',