S0-R1-C5.5-R6-R13 Claim-Bound Numeric Fact Recovery

core/pipeline_agent.py

@@ -7077,16 +7077,27 @@ TEXT:
             r'mechanisms?|groups?|stages?|steps?|waves?|models?|hypotheses?|dimensions?|factors?|items?|'
             r'rounds?|phases?|samples?|studies?|analyses?|conditions?|paths?|constructs?|measures?|components?|'
             r'categories?|levels?|questions?|options?|variables?|sections?|claims?|explanations?|participants?|teams?|'
-            r'orders?|points?|scales?|types?|classes?|periods?|years?|months?|weeks?|days?|trials?|arms?|effects?|generations?'
+            r'orders?|points?|scales?|types?|classes?|periods?|years?|months?|weeks?|days?|trials?|arms?|effects?|generations?|'
+            r'associations?|coefficients?|estimates?|relationships?|correlations?|regressions?|predictors?|outcomes?|mediators?|'
+            r'indicators?|loadings?|intervals?|tests?|diagnostics?'
         )
         found = []
         cardinal_words = '|'.join(cardinal)
-        # Academic attributive compounds are especially important because a
-        # source such as ``eleven-item scale`` naturally becomes ``11项量表``.
-        for match in re.finditer(rf'(?i)\b({cardinal_words})(?:\s+|-)({measurable_nouns})\b', text):
+        # R6-R13: quantity ownership is source-semantic, not surface-adjacent.
+        # Allow a bounded chain of ordinary academic modifiers between the
+        # English number word and a countable/statistical semantic head, e.g.
+        # ``two constituent associations`` or ``two statistically significant
+        # direct paths``. Punctuation cannot be crossed and one source quantity
+        # still authorizes only one matching target numeral occurrence.
+        bounded_modifiers = r'(?:(?:[A-Za-z]+(?:-[A-Za-z]+)?)\s+){0,4}'
+        for match in re.finditer(
+            rf'(?i)\b({cardinal_words})(?:\s+|-){bounded_modifiers}({measurable_nouns})\b', text
+        ):
             found.append(cardinal[match.group(1).casefold()])
         ordinal_words = '|'.join(ordinal)
-        for match in re.finditer(rf'(?i)\b({ordinal_words})(?:\s+|-)({measurable_nouns})\b', text):
+        for match in re.finditer(
+            rf'(?i)\b({ordinal_words})(?:\s+|-){bounded_modifiers}({measurable_nouns})\b', text
+        ):
             found.append(ordinal[match.group(1).casefold()])
         for match in re.finditer(r'(?i)\b(twice|twofold|threefold|fourfold)\b', text):
             found.append({'twice':'2','twofold':'2','threefold':'3','fourfold':'4'}[match.group(1).casefold()])
@@ -8944,6 +8955,45 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
                         )
                         restored[block_id] = translated_with_claims
                         continue
+                if self._is_translation_unowned_numeric_error(error):
+                    claim_match = re.search(r'claim=(C\d{6})', str(error))
+                    source_claims = dict(claim_sources_by_block.get(block_id) or {})
+                    claim_id = claim_match.group(1) if claim_match else ''
+                    if claim_id and claim_id in source_claims:
+                        logging.warning(
+                            'Translation batch numeric claim-local recovery: block=SP-P%s claim=%s error=%s',
+                            block_id, claim_id, self._safe_source_error(error),
+                        )
+                        claim_order = list(source_claims)
+                        _order, segments = self._parse_translation_claim_segments(
+                            parsed[block_id], claim_order
+                        )
+                        sibling_snapshot = {
+                            key: str(value or '') for key, value in segments.items() if key != claim_id
+                        }
+                        segments[claim_id] = self._translate_single_claim_with_recovery(
+                            block_id, claim_id, source_claims[claim_id],
+                            system_prompt, target_lang, ['FACT_INTEGRITY_NUMERIC'],
+                            current_target_claim=segments.get(claim_id),
+                        )
+                        # The failed claim is the only model-owned repair surface.
+                        # Sibling claims must remain byte-stable.
+                        if any(str(segments.get(key) or '') != value for key, value in sibling_snapshot.items()):
+                            raise DeepSeekRequestError(
+                                f'Translation numeric claim-local recovery mutated sibling claim in SP-P{block_id}.'
+                            )
+                        repaired = self._compose_translation_claim_segments(claim_order, segments)
+                        translated_with_claims = self._restore_translation_tokens(
+                            repaired, protected_maps.get(block_id, {}), f'SP-P{block_id}', target_lang,
+                            claim_sources=source_claims, keep_claim_markers=True,
+                            glossary=getattr(self, '_active_translation_glossary', {}),
+                        )
+                        clean_translation = self._strip_translation_claim_markers(translated_with_claims)
+                        self._validate_translation_block_protected(
+                            block_id, source_by_id[block_id], clean_translation, target_lang
+                        )
+                        restored[block_id] = translated_with_claims
+                        continue
                 self._record_translation_batch_damage(target_lang, 'protected_or_claim_integrity')
                 logging.warning(
                     'Translation protected block recovery: block=SP-P%s error=%s; strict isolated retry.',
@@ -9222,17 +9272,23 @@ Each code must be one of: {sorted(allowed)}. Return one or more specific dimensi
         return states
 
     @classmethod
-    def _restore_translation_claim_states(cls, source_blocks, claim_states, section_name=''):
-        """Reassemble marked blocks only when every frozen source claim has a hash-matched durable target."""
+    def _restore_translation_claim_state_segments(cls, source_blocks, claim_states, section_name=''):
+        """Return every hash-valid durable claim independently of missing siblings.
+
+        R6-R13 keeps passed claims reusable even when another claim/block is
+        missing or stale.  This helper never invents a target: it returns only
+        source-hash + semantic-object matched rows and explicitly records the
+        missing claim IDs that still require translation.
+        """
         states = dict(claim_states or {})
-        if not states:
-            return {}, {}, False
-        translations = {}
+        blocks = {}
         pending = {}
+        reusable_claims = 0
         for block_id, source_block in dict(source_blocks or {}).items():
             source_claims = cls._translation_claim_units(source_block)
             order = [item['claim_id'] for item in source_claims]
             segments = {}
+            missing = []
             for item in source_claims:
                 claim_id = item['claim_id']
                 key = f'SP-P{block_id}-{claim_id}'
@@ -9240,26 +9296,90 @@ Each code must be one of: {sorted(allowed)}. Return one or more specific dimensi
                 expected_hash = hashlib.sha256(str(item['text']).encode('utf-8')).hexdigest()
                 expected_semantic_hash = cls._source_semantic_claim_frame(item['text'], section_name).get('frame_hash', '')
                 expected_object_hash = cls._translation_semantic_object_frame(item['text'], section_name).get('object_hash', '')
-                if not isinstance(row, dict) or row.get('source_hash') != expected_hash or not str(row.get('target_text') or '').strip():
-                    return {}, {}, False
-                # R6-R9: durable target claims are semantic-object bound. Any
-                # pre-v1 object state is target-only stale and must be rebuilt;
-                # sealed English remains untouched.
-                if str(row.get('semantic_object_version') or '') != cls.TRANSLATION_SEMANTIC_OBJECT_VERSION:
-                    return {}, {}, False
-                if str(row.get('semantic_object_hash') or '') != expected_object_hash:
-                    return {}, {}, False
-                stored_semantic_hash = str(row.get('semantic_frame_hash') or '')
-                if stored_semantic_hash and stored_semantic_hash != expected_semantic_hash:
-                    return {}, {}, False
+                valid = bool(
+                    isinstance(row, dict)
+                    and row.get('source_hash') == expected_hash
+                    and str(row.get('target_text') or '').strip()
+                    and str(row.get('semantic_object_version') or '') == cls.TRANSLATION_SEMANTIC_OBJECT_VERSION
+                    and str(row.get('semantic_object_hash') or '') == expected_object_hash
+                )
+                if valid:
+                    stored_semantic_hash = str(row.get('semantic_frame_hash') or '')
+                    if stored_semantic_hash and stored_semantic_hash != expected_semantic_hash:
+                        valid = False
+                if not valid:
+                    missing.append(claim_id)
+                    continue
                 segments[claim_id] = str(row.get('target_text') or '').strip()
+                reusable_claims += 1
                 if str(row.get('qa_status') or '') == 'repair_pending':
-                    codes = [str(code or '').strip().upper() for code in (row.get('issue_codes') or []) if str(code or '').strip()]
+                    codes = [
+                        str(code or '').strip().upper() for code in (row.get('issue_codes') or [])
+                        if str(code or '').strip()
+                    ]
                     if codes:
                         pending.setdefault(str(block_id), {})[str(claim_id)] = codes[:5]
-            translations[str(block_id)] = cls._compose_translation_claim_segments(order, segments)
+            blocks[str(block_id)] = {
+                'order': order, 'segments': segments, 'missing': missing,
+                'source_claims': {item['claim_id']: item['text'] for item in source_claims},
+            }
+        return blocks, pending, reusable_claims
+
+    @classmethod
+    def _restore_translation_claim_states(cls, source_blocks, claim_states, section_name=''):
+        """Compatibility wrapper: full resume succeeds only when every claim is durable."""
+        if not dict(claim_states or {}):
+            return {}, {}, False
+        blocks, pending, reusable_claims = cls._restore_translation_claim_state_segments(
+            source_blocks, claim_states, section_name=section_name
+        )
+        if reusable_claims <= 0 or any(row.get('missing') for row in blocks.values()):
+            return {}, {}, False
+        translations = {
+            str(block_id): cls._compose_translation_claim_segments(row['order'], row['segments'])
+            for block_id, row in blocks.items()
+        }
         return translations, pending, True
 
+    def _audit_translation_claim_subset_with_retry(
+        self, source_blocks, translations, target_lang, glossary, claim_pairs, section_name=''
+    ):
+        """QA only newly translated claims; durable passed siblings never re-enter the model."""
+        synthetic_sources = {}
+        synthetic_targets = {}
+        reverse = {}
+        for index, (block_id, claim_id) in enumerate(claim_pairs, start=1):
+            source_claims = {
+                item['claim_id']: item['text'] for item in self._translation_claim_units(source_blocks[block_id])
+            }
+            order = list(source_claims)
+            _order, target_claims = self._parse_translation_claim_segments(translations[block_id], order)
+            synthetic_id = f'{900000 + index:06d}'
+            synthetic_sources[synthetic_id] = source_claims[claim_id]
+            synthetic_targets[synthetic_id] = self._compose_translation_claim_segments(
+                ['C000001'], {'C000001': target_claims[claim_id]}
+            )
+            reverse[(synthetic_id, 'C000001')] = (str(block_id), str(claim_id))
+        if not synthetic_sources:
+            return {}
+        synthetic_issues = self._audit_translation_blocks_with_retry(
+            synthetic_sources, synthetic_targets, target_lang, glossary,
+            section_name=section_name or 'frozen section', max_attempts=2,
+        )
+        issues = {}
+        for synthetic_id, claim_map in dict(synthetic_issues or {}).items():
+            for synthetic_claim_id, codes in dict(claim_map or {}).items():
+                original = reverse.get((str(synthetic_id), str(synthetic_claim_id)))
+                if not original:
+                    continue
+                block_id, claim_id = original
+                synthetic_evidence_key = (str(synthetic_id), str(synthetic_claim_id))
+                evidence = dict(self._translation_claim_qa_evidence.pop(synthetic_evidence_key, {}) or {})
+                if evidence:
+                    self._translation_claim_qa_evidence[(block_id, claim_id)] = evidence
+                issues.setdefault(block_id, {})[claim_id] = list(codes or [])
+        return issues
+
     def _adjudicate_persistent_translation_issues(
         self, sources, translations, issues, target_lang, glossary=None, section_name=''
     ):
@@ -9738,20 +9858,62 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
                         sum(len(v or {}) for v in quality_issues.values()),
                     )
                 else:
-                    pending = list(pending_blocks)
-                    while pending:
-                        batch, pending = self._take_adaptive_translation_batch(
-                            pending, target_lang, token_budget
-                        )
-                        damage_before = self._translation_batch_damage_score(target_lang)
-                        translations.update(self._translate_marked_batch_with_recovery(batch, system_prompt, target_lang))
-                        if self._translation_batch_damage_score(target_lang) == damage_before and damage_before > 0:
-                            self._relax_translation_batch_damage(target_lang)
-                    emit_stage('qa')
-                    quality_issues = self._audit_translation_blocks_with_retry(
-                        source_blocks, translations, target_lang, frozen_glossary,
-                        section_name=section_name or 'frozen section', max_attempts=2,
+                    partial_blocks, partial_pending, reusable_claims = self._restore_translation_claim_state_segments(
+                        source_blocks, claim_states, section_name=section_name or 'frozen section'
                     )
+                    if reusable_claims > 0:
+                        translations = {}
+                        quality_issues = {
+                            str(block_id): {str(claim_id): list(codes or []) for claim_id, codes in dict(claim_map or {}).items()}
+                            for block_id, claim_map in dict(partial_pending or {}).items()
+                        }
+                        newly_translated_pairs = []
+                        for block_id, row in partial_blocks.items():
+                            segments = dict(row.get('segments') or {})
+                            source_claims = dict(row.get('source_claims') or {})
+                            for claim_id in list(row.get('missing') or []):
+                                segments[claim_id] = self._translate_single_claim_with_recovery(
+                                    block_id, claim_id, source_claims[claim_id], system_prompt, target_lang,
+                                    ['DURABLE_RESUME_MISSING'], repair_strategy='SOURCE_REGENERATE',
+                                )
+                                newly_translated_pairs.append((block_id, claim_id))
+                            translations[block_id] = self._compose_translation_claim_segments(row['order'], segments)
+                        if newly_translated_pairs:
+                            emit_stage('qa')
+                            fresh_issues = self._audit_translation_claim_subset_with_retry(
+                                source_blocks, translations, target_lang, frozen_glossary,
+                                newly_translated_pairs, section_name=section_name or 'frozen section',
+                            )
+                            for block_id, claim_map in dict(fresh_issues or {}).items():
+                                for claim_id, codes in dict(claim_map or {}).items():
+                                    quality_issues.setdefault(block_id, {})[claim_id] = list(codes or [])
+                        logging.info(
+                            'Translation partial durable claim resume | section=%s reused=%s translated_missing=%s pending=%s',
+                            section_name or 'section', reusable_claims, len(newly_translated_pairs),
+                            sum(len(v or {}) for v in quality_issues.values()),
+                        )
+                    else:
+                        pending = list(pending_blocks)
+                        while pending:
+                            batch, pending = self._take_adaptive_translation_batch(
+                                pending, target_lang, token_budget
+                            )
+                            damage_before = self._translation_batch_damage_score(target_lang)
+                            translations.update(self._translate_marked_batch_with_recovery(batch, system_prompt, target_lang))
+                            # R6-R13: persist every completed claim batch immediately.
+                            # If a later block fails deterministically, already-passed
+                            # claims survive the runner's repair_pending checkpoint.
+                            self._last_translation_claim_states = self._snapshot_translation_claim_states(
+                                source_blocks, translations, {}, self._translation_claim_qa_evidence,
+                                self._translation_claim_repair_meta, section_name=section_name or 'frozen section',
+                            )
+                            if self._translation_batch_damage_score(target_lang) == damage_before and damage_before > 0:
+                                self._relax_translation_batch_damage(target_lang)
+                        emit_stage('qa')
+                        quality_issues = self._audit_translation_blocks_with_retry(
+                            source_blocks, translations, target_lang, frozen_glossary,
+                            section_name=section_name or 'frozen section', max_attempts=2,
+                        )
                 self._last_translation_claim_states = self._snapshot_translation_claim_states(
                     source_blocks, translations, quality_issues, self._translation_claim_qa_evidence,
                     self._translation_claim_repair_meta, section_name=section_name or 'frozen section',