S0-R1-C5.5-R6-R3 Cross-Mode Four-Language Repair Entitlement Closure

core/generation_runner.py

@@ -1082,7 +1082,10 @@ class GenerationRunner:
         lineages = {cls._section_violation_lineage(row) for row in (rows or [])}
         exhausted = {name for name in lineages if int(lineage_attempts.get(name, 0)) >= 1}
         fresh = lineages - exhausted
-        return bool(fresh) and not exhausted and not hard_cap_reached, exhausted, fresh
+        # R6-R3 fairness closure: an exhausted lineage may not steal the bounded
+        # repair entitlement of a genuinely fresh lineage discovered in the same
+        # validation round.  The Section hard cap remains the global stop.
+        return bool(fresh) and not hard_cap_reached, exhausted, fresh
 
     @classmethod
     def _partition_section_violations(cls, rows, *, section_name: str = "", mode: str = "", exhausted: bool = False):
@@ -1097,6 +1100,30 @@ class GenerationRunner:
                 blocking.append(row)
         return blocking, advisory
 
+    @classmethod
+    def _partition_section_violations_by_lineage(
+        cls, rows, lineage_attempts, *, section_name: str = "", mode: str = "",
+        hard_cap_reached: bool = False,
+    ):
+        """Classify each defect using only its own repair-exhaustion state.
+
+        R6-R3 prevents a previously exhausted sibling lineage from globally
+        changing the severity of a newly discovered defect.  When the Section
+        hard cap is reached every lineage is evaluated as exhausted.
+        """
+        blocking, advisory = [], []
+        for row in list(rows or []):
+            lineage = cls._section_violation_lineage(row)
+            lineage_exhausted = bool(hard_cap_reached) or int(lineage_attempts.get(lineage, 0)) >= 1
+            severity = cls._section_violation_severity(
+                row, section_name=section_name, mode=mode, exhausted=lineage_exhausted
+            )
+            if severity == "ADVISORY":
+                advisory.append(row)
+            else:
+                blocking.append(row)
+        return blocking, advisory
+
     @staticmethod
     def _record_section_advisories(quality_notes: list, rows, *, section_name: str = "") -> None:
         """Append reader-facing, nonfatal notes without exposing control vocabulary."""
@@ -3920,23 +3947,27 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     may_retry, exhausted_lineages, fresh_lineages = self._section_retry_decision(
                         preview_blocking, lineage_attempts, hard_cap_reached=hard_cap_reached
                     )
-                    exhausted = hard_cap_reached or bool(exhausted_lineages)
-                    blocking_rows, advisory_rows = self._partition_section_violations(
-                        phase1_validation_rows, section_name=display_name, mode=mode, exhausted=exhausted
-                    )
-                    if blocking_rows and may_retry:
+                    if may_retry:
+                        retry_rows = [
+                            row for row in preview_blocking
+                            if self._section_violation_lineage(row) in fresh_lineages
+                        ]
                         for lineage in fresh_lineages:
                             lineage_attempts[lineage] = int(lineage_attempts.get(lineage, 0)) + 1
-                        attempt_feedback = self._section_retry_feedback(blocking_rows)
+                        attempt_feedback = self._section_retry_feedback(retry_rows)
                         logging.warning(
-                            "Unified Phase-1 Section lineage retry | unit=%s attempt=%s lineages=%s count=%s types=%s",
-                            display_name, section_attempt, sorted(fresh_lineages), len(blocking_rows),
-                            sorted({str(row.get("type") or "") for row in blocking_rows if isinstance(row, dict)}),
+                            "Unified Phase-1 Section lineage retry | unit=%s attempt=%s lineages=%s exhausted_siblings=%s count=%s types=%s",
+                            display_name, section_attempt, sorted(fresh_lineages), sorted(exhausted_lineages), len(retry_rows),
+                            sorted({str(row.get("type") or "") for row in retry_rows if isinstance(row, dict)}),
                         )
                         self._restore_section_citation_state(
                             agent, citation_plan, display_name, frozen_citation_state
                         )
                         continue
+                    blocking_rows, advisory_rows = self._partition_section_violations_by_lineage(
+                        phase1_validation_rows, lineage_attempts, section_name=display_name, mode=mode,
+                        hard_cap_reached=hard_cap_reached
+                    )
                     if advisory_rows:
                         self._record_section_advisories(
                             quality_notes, advisory_rows, section_name=display_name
@@ -4104,23 +4135,27 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     may_retry, exhausted_lineages, fresh_lineages = self._section_retry_decision(
                         preview_blocking, lineage_attempts, hard_cap_reached=hard_cap_reached
                     )
-                    exhausted = hard_cap_reached or bool(exhausted_lineages)
-                    blocking_rows, advisory_rows = self._partition_section_violations(
-                        post_validation_rows, section_name=display_name, mode=mode, exhausted=exhausted
-                    )
-                    if blocking_rows and may_retry:
+                    if may_retry:
+                        retry_rows = [
+                            row for row in preview_blocking
+                            if self._section_violation_lineage(row) in fresh_lineages
+                        ]
                         for lineage in fresh_lineages:
                             lineage_attempts[lineage] = int(lineage_attempts.get(lineage, 0)) + 1
-                        attempt_feedback = self._section_retry_feedback(blocking_rows)
+                        attempt_feedback = self._section_retry_feedback(retry_rows)
                         logging.warning(
-                            "Post-normalization Section lineage retry | unit=%s attempt=%s lineages=%s count=%s types=%s",
-                            display_name, section_attempt, sorted(fresh_lineages), len(blocking_rows),
-                            sorted({str(row.get("type") or "") for row in blocking_rows if isinstance(row, dict)}),
+                            "Post-normalization Section lineage retry | unit=%s attempt=%s lineages=%s exhausted_siblings=%s count=%s types=%s",
+                            display_name, section_attempt, sorted(fresh_lineages), sorted(exhausted_lineages), len(retry_rows),
+                            sorted({str(row.get("type") or "") for row in retry_rows if isinstance(row, dict)}),
                         )
                         self._restore_section_citation_state(
                             agent, citation_plan, display_name, frozen_citation_state
                         )
                         continue
+                    blocking_rows, advisory_rows = self._partition_section_violations_by_lineage(
+                        post_validation_rows, lineage_attempts, section_name=display_name, mode=mode,
+                        hard_cap_reached=hard_cap_reached
+                    )
                     if advisory_rows:
                         self._record_section_advisories(
                             quality_notes, advisory_rows, section_name=display_name

core/pipeline_agent.py

@@ -8219,6 +8219,20 @@ Each code must be one of: {sorted(allowed)}. Return one or more specific dimensi
                     'last_evidence_signature': str(meta.get('last_evidence_signature') or ''),
                     'last_strategy': str(meta.get('last_strategy') or ''),
                     'nonconvergent_state': str(meta.get('nonconvergent_state') or ''),
+                    # R6-R3: persist defect-family ownership across durable resume.
+                    # Without these fields a restarted RU/KK/ZH claim can collapse
+                    # back to legacy claim-level history and either revive an
+                    # exhausted family or deny a genuinely fresh family its turn.
+                    'family_attempt_counts': {
+                        str(family): int(count or 0)
+                        for family, count in dict(meta.get('family_attempt_counts') or {}).items()
+                        if str(family or '').strip()
+                    },
+                    'family_last_strategy': {
+                        str(family): str(strategy or '').strip().upper()
+                        for family, strategy in dict(meta.get('family_last_strategy') or {}).items()
+                        if str(family or '').strip()
+                    },
                 }
         return states
 
@@ -8736,7 +8750,8 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
                         current_target_claim = segments[claim_id]
                         evidence_key = (block_id, claim_id)
                         repair_evidence = dict(self._translation_claim_qa_evidence.get(evidence_key) or {})
-                        code_set = {str(code or '').strip().upper() for code in (codes or [])}
+                        observed_code_set = {str(code or '').strip().upper() for code in (codes or [])}
+                        code_set = set(observed_code_set)
                         meta = self._translation_claim_repair_meta.setdefault(evidence_key, {
                             'repair_attempt_count': 0, 'last_issue_codes': [], 'last_evidence_signature': '',
                             'last_strategy': '', 'nonconvergent_state': '',
@@ -8744,27 +8759,48 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
                         })
                         prior_attempts = int(meta.get('repair_attempt_count') or 0)
                         prior_codes = set(meta.get('last_issue_codes') or [])
-                        families = self._translation_defect_family(code_set)
+                        observed_families = self._translation_defect_family(code_set)
                         family_counts = dict(meta.get('family_attempt_counts') or {})
                         family_strategies = dict(meta.get('family_last_strategy') or {})
                         exhausted_families = {
-                            family for family in families
+                            family for family in observed_families
                             if int(family_counts.get(family, 0)) >= self.MAX_TRANSLATION_CLAIM_REPAIR_ATTEMPTS
                         }
+                        eligible_families = set(observed_families) - exhausted_families
                         claim_hard_cap = prior_attempts >= self.MAX_TRANSLATION_CLAIM_TOTAL_REPAIRS
-                        if exhausted_families or claim_hard_cap:
+                        if claim_hard_cap or not eligible_families:
                             meta['nonconvergent_state'] = 'exhausted_family:' + ','.join(sorted(exhausted_families)) if exhausted_families else 'exhausted_after_claim_hard_cap'
                             carried_exhausted.setdefault(block_id, {})[claim_id] = list(codes)
                             logging.info(
                                 'Translation claim defect-family entitlement exhausted: section=%s id=SP-P%s-%s attempts=%s families=%s',
-                                section_name or 'section', block_id, claim_id, prior_attempts, sorted(exhausted_families or set(families)),
+                                section_name or 'section', block_id, claim_id, prior_attempts, sorted(exhausted_families or set(observed_families)),
                             )
                             self._record_translation_assurance(
                                 'CLAIM_REPAIR_EXHAUSTED', target_lang=target_lang,
                                 block_id=f'SP-P{block_id}-{claim_id}', severity='critical', decision='ADJUDICATE',
-                                detail={'attempts': prior_attempts, 'families': '|'.join(sorted(exhausted_families or set(families)))},
+                                detail={'attempts': prior_attempts, 'families': '|'.join(sorted(exhausted_families or set(observed_families)))},
                             )
                             continue
+                        if exhausted_families:
+                            # R6-R3 fairness closure: do not let an exhausted sibling
+                            # family consume or block the entitlement of a still-eligible
+                            # family.  Repair only codes owned by eligible families; the
+                            # exhausted defect remains subject to later adjudication if it
+                            # survives the fresh-family repair.
+                            active_codes = [
+                                str(code or '').strip().upper() for code in (codes or [])
+                                if set(self._translation_defect_family([code])) & eligible_families
+                            ]
+                            codes = list(dict.fromkeys(active_codes))
+                            code_set = set(codes)
+                            families = self._translation_defect_family(code_set)
+                            logging.info(
+                                'Translation mixed exhausted+eligible families: section=%s id=SP-P%s-%s exhausted=%s eligible=%s repair_codes=%s',
+                                section_name or 'section', block_id, claim_id, sorted(exhausted_families),
+                                sorted(eligible_families), sorted(code_set),
+                            )
+                        else:
+                            families = observed_families
                         if prior_attempts == 0 and repair_cycle > 1:
                             logging.info(
                                 'Translation late-detected claim receives own repair entitlement: section=%s id=SP-P%s-%s cycle=%s',