S0-R1-C5.5-R4 Cross-Mode Defect-Lineage Retry Ownership + Methodological Causal-Scope Closure

core/generation_runner.py

@@ -1035,6 +1035,48 @@ class GenerationRunner:
         # new to the severity registry.
         return "FATAL"
 
+    @staticmethod
+    def _section_violation_lineage(row) -> str:
+        """Return a stable repair lineage for one section validation defect.
+
+        R4 assigns the bounded rewrite entitlement to the defect lineage rather
+        than to the whole Section.  Closely related validators share a lineage
+        so spelling variants of the same integrity problem cannot manufacture
+        fresh retries.
+        """
+        item = dict(row or {}) if isinstance(row, dict) else {"type": "unknown_validation_issue"}
+        kind = str(item.get("type") or "unknown_validation_issue")
+        groups = {
+            "citation": {"citation_claim_split_required", "citation_unsupported_claim", "citation_support_unverified",
+                         "citation_outside_frozen_section_whitelist"},
+            "causal_scope": {"causal_language_violation"},
+            "quant_method": {"unsupported_quantitative_method", "quantitative_metric_mismatch", "unfrozen_p_value"},
+            "study_fact": {"sample_size_mismatch", "sample_count_mismatch", "scale_points_mismatch",
+                           "construct_role_mismatch", "actor_data_source_relation_mismatch", "unsupported_study_procedure"},
+            "enumeration": {"enumeration_contract_invalid", "enumeration_contract_mismatch"},
+            "hypothesis": {"hypothesis_contract_missing_id", "hypothesis_contract_ordinal_alias",
+                           "hypothesis_contract_pairing_mismatch", "hypothesis_contract_unknown_id"},
+            "workflow_trace": {"authorial_ai_meta_trace", "author_facing_internal_workflow_language"},
+        }
+        for lineage, kinds in groups.items():
+            if kind in kinds:
+                return lineage
+        return kind
+
+    @classmethod
+    def _section_retry_decision(cls, rows, lineage_attempts, *, hard_cap_reached: bool = False):
+        """Return (may_retry, exhausted_lineages, new_lineages).
+
+        Each lineage receives one rewrite entitlement.  A lineage that survives
+        its own rewrite remains fail-closed.  A genuinely new lineage discovered
+        only after a previous repair receives one bounded chance, subject to the
+        Section hard cap.
+        """
+        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
+
     @classmethod
     def _partition_section_violations(cls, rows, *, section_name: str = "", mode: str = "", exhausted: bool = False):
         blocking, advisory = [], []
@@ -3621,7 +3663,10 @@ Output only final academic prose. Do not report the word count."""
                 raise RuntimeError(f"section_retry_exhausted_requires_new_lifecycle:{display_name}")
 
             deterministic_appendix = mode == "asq" and section_id == "appendix"
-            max_section_attempts = 1 if deterministic_appendix else 2
+            # R4: one initial attempt plus bounded lineage-owned repair turns.
+            # Four is a hard safety cap, not a blanket four-attempt entitlement.
+            max_section_attempts = 1 if deterministic_appendix else 4
+            lineage_attempts = {}
             frozen_citation_state = self._snapshot_section_citation_state(
                 agent, citation_plan, display_name
             )
@@ -3752,19 +3797,27 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                 )
                 if phase1_validation_rows:
                     retry_excluded_tags.update(self._citation_violation_tags(citation_semantic_rows))
-                    exhausted = section_attempt >= max_section_attempts
+                    hard_cap_reached = section_attempt >= max_section_attempts
+                    # Severity is evaluated as exhausted only when this defect
+                    # lineage has already consumed its own repair entitlement.
+                    preview_blocking, preview_advisory = self._partition_section_violations(
+                        phase1_validation_rows, section_name=display_name, mode=mode, exhausted=False
+                    )
+                    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 not exhausted:
+                    if blocking_rows and may_retry:
+                        for lineage in fresh_lineages:
+                            lineage_attempts[lineage] = int(lineage_attempts.get(lineage, 0)) + 1
                         attempt_feedback = self._section_retry_feedback(blocking_rows)
                         logging.warning(
-                            "Unified Phase-1 Section retry | unit=%s attempt=%s count=%s types=%s",
-                            display_name, section_attempt, 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 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)}),
                         )
                         self._restore_section_citation_state(
                             agent, citation_plan, display_name, frozen_citation_state
@@ -3930,15 +3983,24 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     hypothesis_ledger=hypothesis_ledger,
                 )
                 if post_validation_rows:
-                    exhausted = section_attempt >= max_section_attempts
+                    hard_cap_reached = section_attempt >= max_section_attempts
+                    preview_blocking, preview_advisory = self._partition_section_violations(
+                        post_validation_rows, section_name=display_name, mode=mode, exhausted=False
+                    )
+                    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 not exhausted:
+                    if blocking_rows and may_retry:
+                        for lineage in fresh_lineages:
+                            lineage_attempts[lineage] = int(lineage_attempts.get(lineage, 0)) + 1
                         attempt_feedback = self._section_retry_feedback(blocking_rows)
                         logging.warning(
-                            "Post-normalization Section retry | unit=%s attempt=%s count=%s types=%s",
-                            display_name, section_attempt, len(blocking_rows),
+                            "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)}),
                         )
                         self._restore_section_citation_state(

core/pipeline_agent.py

@@ -72,9 +72,33 @@ class CitationAuditError(DeepSeekRequestError):
 class SSciPipelineAgent:
     TRANSLATION_CONTRACT_VERSION = 'hierarchical-multilingual-v8'
     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-v1'
 
+    @staticmethod
+    def _translation_defect_family(codes):
+        """Collapse translation QA codes into stable repair lineages.
+
+        R4 gives each claim+defect family its own bounded entitlement while
+        retaining a claim-level hard cap. Closely related semantic codes share
+        one family so re-labeling the same defect cannot reset the budget.
+        """
+        normalized = {str(code or '').strip().upper() for code in (codes or []) if str(code or '').strip()}
+        families = []
+        if normalized & {'SEMANTIC_DRIFT','NEGATION_DRIFT','CAUSALITY_STRENGTH_DRIFT','UNCERTAINTY_DRIFT','OMITTED_CLAIM','ADDED_CLAIM'}:
+            families.append('SEMANTIC')
+        if 'GRAMMAR' in normalized:
+            families.append('GRAMMAR')
+        if 'TERMINOLOGY' in normalized:
+            families.append('TERMINOLOGY')
+        if 'LANGUAGE_CONTAMINATION' in normalized:
+            families.append('LANGUAGE_CONTAMINATION')
+        for code in sorted(normalized):
+            if code not in {'SEMANTIC_DRIFT','NEGATION_DRIFT','CAUSALITY_STRENGTH_DRIFT','UNCERTAINTY_DRIFT','OMITTED_CLAIM','ADDED_CLAIM','GRAMMAR','TERMINOLOGY','LANGUAGE_CONTAMINATION'}:
+                families.append(code)
+        return tuple(dict.fromkeys(families)) or ('UNKNOWN',)
+
     def __init__(self, ds_key, s2_key=""):
         self.ds_key = str(ds_key or "").strip()
         if not self.ds_key:
@@ -8268,12 +8292,26 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
             claim_key = str(row.get('claim_id') or '')
             if not block_key or not claim_key:
                 continue
+            prior_codes = [str(code or '').strip().upper() for code in (row.get('last_issue_codes') or []) if str(code or '').strip()][:5]
+            family_attempt_counts = dict(row.get('family_attempt_counts') or {})
+            family_last_strategy = dict(row.get('family_last_strategy') or {})
+            legacy_state = str(row.get('nonconvergent_state') or '')
+            legacy_attempts = int(row.get('repair_attempt_count') or 0)
+            # R4 migration rule: an R3 checkpoint that had already exhausted a
+            # claim/family remains fail-closed. New lineage ownership must never
+            # resurrect a previously terminal defect.
+            if not family_attempt_counts and (legacy_state.startswith('exhausted') or legacy_attempts >= self.MAX_TRANSLATION_CLAIM_REPAIR_ATTEMPTS):
+                for family in self._translation_defect_family(prior_codes):
+                    family_attempt_counts[family] = self.MAX_TRANSLATION_CLAIM_REPAIR_ATTEMPTS
+                    family_last_strategy[family] = str(row.get('last_strategy') or '').strip().upper()
             self._translation_claim_repair_meta[(block_key, claim_key)] = {
-                'repair_attempt_count': int(row.get('repair_attempt_count') or 0),
-                'last_issue_codes': [str(code or '').strip().upper() for code in (row.get('last_issue_codes') or []) if str(code or '').strip()][:5],
+                'repair_attempt_count': legacy_attempts,
+                'last_issue_codes': prior_codes,
                 'last_evidence_signature': str(row.get('last_evidence_signature') or ''),
                 'last_strategy': str(row.get('last_strategy') or '').strip().upper(),
-                'nonconvergent_state': str(row.get('nonconvergent_state') or ''),
+                'nonconvergent_state': legacy_state,
+                'family_attempt_counts': family_attempt_counts,
+                'family_last_strategy': family_last_strategy,
             }
 
         def emit_stage(stage):
@@ -8330,7 +8368,7 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
                 )
                 repair_cycle = 0
                 total_claims = sum(len(self._translation_claim_units(source)) for source in source_blocks.values())
-                max_repair_cycles = max(1, total_claims * self.MAX_TRANSLATION_CLAIM_REPAIR_ATTEMPTS)
+                max_repair_cycles = max(1, total_claims * self.MAX_TRANSLATION_CLAIM_TOTAL_REPAIRS)
                 while quality_issues and repair_cycle < max_repair_cycles:
                     repair_cycle += 1
                     failed_pairs = [
@@ -8366,26 +8404,29 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
                         meta = self._translation_claim_repair_meta.setdefault(evidence_key, {
                             'repair_attempt_count': 0, 'last_issue_codes': [], 'last_evidence_signature': '',
                             'last_strategy': '', 'nonconvergent_state': '',
+                            'family_attempt_counts': {}, 'family_last_strategy': {},
                         })
                         prior_attempts = int(meta.get('repair_attempt_count') or 0)
                         prior_codes = set(meta.get('last_issue_codes') or [])
-                        already_source_regenerated = str(meta.get('last_strategy') or '').upper() == 'SOURCE_REGENERATE'
-                        already_exhausted = str(meta.get('nonconvergent_state') or '').startswith('exhausted')
-                        if (
-                            prior_attempts >= self.MAX_TRANSLATION_CLAIM_REPAIR_ATTEMPTS
-                            or already_exhausted
-                            or (already_source_regenerated and bool(prior_codes & code_set))
-                        ):
-                            meta['nonconvergent_state'] = 'exhausted_after_source_regenerate' if already_source_regenerated else 'exhausted_after_claim_budget'
+                        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
+                            if int(family_counts.get(family, 0)) >= self.MAX_TRANSLATION_CLAIM_REPAIR_ATTEMPTS
+                        }
+                        claim_hard_cap = prior_attempts >= self.MAX_TRANSLATION_CLAIM_TOTAL_REPAIRS
+                        if exhausted_families or claim_hard_cap:
+                            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 repair entitlement exhausted: section=%s id=SP-P%s-%s attempts=%s strategy=%s',
-                                section_name or 'section', block_id, claim_id, prior_attempts, meta.get('last_strategy') or '',
+                                '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)),
                             )
                             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, 'strategy': str(meta.get('last_strategy') or '')},
+                                detail={'attempts': prior_attempts, 'families': '|'.join(sorted(exhausted_families or set(families)))},
                             )
                             continue
                         if prior_attempts == 0 and repair_cycle > 1:
@@ -8485,9 +8526,15 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
                             'UNRESOLVED_SEMANTIC_DELTA' in set(repair_evidence.get('semantic_delta_codes') or [])
                             or 'GRAMMAR_UNSPECIFIED' in set(repair_evidence.get('grammar_evidence_codes') or [])
                         )
+                        family_prior_attempts = max(int(family_counts.get(family, 0)) for family in families)
+                        family_prior_source_regenerated = any(
+                            str(family_strategies.get(family) or '').upper() == 'SOURCE_REGENERATE'
+                            for family in families
+                        )
                         strategy = 'SOURCE_REGENERATE' if (
                             force_source_regenerate
-                            or prior_attempts >= 1
+                            or family_prior_attempts >= 1
+                            or family_prior_source_regenerated
                             or (same_defect and str(meta.get('last_strategy') or '').upper() == 'LOCAL_EDIT')
                         ) else 'LOCAL_EDIT'
                         if strategy == 'SOURCE_REGENERATE':
@@ -8517,10 +8564,17 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
                             **repair_kwargs,
                         )
                         meta['repair_attempt_count'] = int(meta.get('repair_attempt_count') or 0) + 1
+                        family_counts = dict(meta.get('family_attempt_counts') or family_counts)
+                        family_strategies = dict(meta.get('family_last_strategy') or family_strategies)
+                        for family in families:
+                            family_counts[family] = int(family_counts.get(family, 0)) + 1
+                            family_strategies[family] = strategy
+                        meta['family_attempt_counts'] = family_counts
+                        meta['family_last_strategy'] = family_strategies
                         meta['last_issue_codes'] = sorted(code_set)[:5]
                         meta['last_evidence_signature'] = signature
                         meta['last_strategy'] = strategy
-                        meta['nonconvergent_state'] = 'source_regenerated' if strategy == 'SOURCE_REGENERATE' else ''
+                        meta['nonconvergent_state'] = ''
                         if 'LANGUAGE_CONTAMINATION' in code_set:
                             post_repair_lexical = self._translation_lexical_purity_evidence(
                                 source_claims[claim_id], repaired_claim, target_lang, frozen_glossary

core/quantitative_engine.py

@@ -1091,6 +1091,47 @@ Keywords: {keywords}"""
         """Compatibility wrapper for older call sites; uses the shared tri-state contract."""
         return cls.classify_quantitative_causal_sentence(sentence) == "EXPLICIT_CAUSAL_LIMITATION"
 
+    @classmethod
+    def _is_methodological_bias_mechanism_sentence(cls, sentence, known_constructs=None):
+        """Recognize measurement/method-bias mechanisms without weakening construct causality.
+
+        R4 separates methodological statements about how measurement conditions
+        may bias statistical quantities from substantive claims that one study
+        construct causes another.  The exemption is intentionally narrow: a
+        measurement/source-bias actor and a statistical/measurement target must
+        both be present.  Explicit construct-level causal language still wins.
+        """
+        value = re.sub(r"\s+", " ", str(sentence or "")).strip().casefold()
+        if not value:
+            return False
+        methodological_actor = re.search(
+            r"\b(?:self[- ]reports?|single\s+questionnaire|same[- ]source|common\s+method(?:\s+bias)?|"
+            r"response\s+styles?|social\s+desirability|measurement\s+error|item\s+wording|"
+            r"common\s+rater|method\s+variance|single[- ]source\s+measurement)\b", value, re.I
+        )
+        statistical_target = re.search(
+            r"\b(?:correlations?|covariances?|associations?|estimates?|coefficients?|variance|"
+            r"standard\s+errors?|reliability|validity|measurement\s+error|method\s+variance)\b", value, re.I
+        )
+        mechanism = re.search(
+            r"\b(?:inflate|inflates|inflated|attenuate|attenuates|attenuated|bias|biases|biased|"
+            r"distort|distorts|distorted|increase|increases|increased|decrease|decreases|decreased|"
+            r"contribute|contributes|contributed|induce|induces|induced)\b", value, re.I
+        )
+        if not (methodological_actor and statistical_target and mechanism):
+            return False
+
+        # Never let a methodological phrase shield an explicit relation between
+        # named study constructs.
+        constructs = [str(x or '').strip().casefold() for x in (known_constructs or []) if str(x or '').strip()]
+        named_hits = [c for c in constructs if c and c in value]
+        if len(named_hits) >= 2 and re.search(
+            r"\b(?:causes?|leads?\s+to|drives?|determines?|produces?|results?\s+in|yields?|generates?)\b",
+            value, re.I
+        ):
+            return False
+        return True
+
     @classmethod
     def inspect_quantitative_study_fact_contract(cls, text, metadata, section_name=""):
         """Pure AMJ quantitative Evidence Ledger inspection; never rewrites prose."""
@@ -1162,7 +1203,13 @@ Keywords: {keywords}"""
                     if pattern.search(sentence):
                         violations.append({"type":"construct_role_mismatch","role":role,"observed":candidate,"expected":expected,"preview":sentence.strip()[:180]})
             if strict:
-                causal_state = cls.classify_quantitative_causal_sentence(sentence)
+                methodological_bias_scope = cls._is_methodological_bias_mechanism_sentence(
+                    sentence, known_constructs
+                )
+                causal_state = (
+                    "NON_CAUSAL" if methodological_bias_scope
+                    else cls.classify_quantitative_causal_sentence(sentence)
+                )
                 if causal_state == "AFFIRMATIVE_CAUSAL":
                     violations.append({
                         "type":"causal_language_violation",