P0-R1.1-R2-R1 Evidence-Sealed Finalization

core/generation_runner.py

@@ -45,6 +45,7 @@ CITATION_ROLE_RULES = {
 }
 TRANSLATION_CONTRACT_VERSION = 'hierarchical-multilingual-v8'
 FINAL_CITATION_LEDGER_VERSION = 'resolved-final-ledger-v1'
+ENGLISH_SECTION_SEAL_VERSION = 'english-section-integrity-seal-v1'
 
 
 def _clean_trailing_references(text, keywords=("reference", "references", "reference list", "参考文献")):
@@ -929,6 +930,43 @@ class GenerationRunner:
             )
         return value, list(diagnostics or [])
 
+    @staticmethod
+    def _section_integrity_hash(value: str) -> str:
+        exact = str(value or "").replace("\r\n", "\n").replace("\r", "\n").strip()
+        return hashlib.sha256(exact.encode("utf-8")).hexdigest()
+
+    @classmethod
+    def _build_section_integrity_seal(cls, section_key: str, section_title: str, section_text: str) -> dict:
+        return {
+            "version": ENGLISH_SECTION_SEAL_VERSION,
+            "section_key": str(section_key or ""),
+            "section_title": str(section_title or ""),
+            "sha256": cls._section_integrity_hash(section_text),
+        }
+
+    @staticmethod
+    def _replace_top_level_section(manuscript: str, title: str, replacement: str) -> str:
+        value = str(manuscript or "")
+        pattern = re.compile(
+            rf'(?ms)^#\s+{re.escape(str(title))}\s*$\n.*?(?=^#\s+[^#].*?$|\Z)'
+        )
+        matches = list(pattern.finditer(value))
+        if len(matches) != 1:
+            raise RuntimeError(f"top_level_section_replace_count_invalid:{title}:{len(matches)}")
+        match = matches[0]
+        rendered = str(replacement or "").strip()
+        return (value[:match.start()] + rendered + "\n\n" + value[match.end():].lstrip("\n")).strip() + "\n\n"
+
+    @classmethod
+    def _validate_section_integrity_seal(cls, seal: dict, section_text: str, section_title: str) -> None:
+        record = dict(seal or {})
+        if record.get("version") != ENGLISH_SECTION_SEAL_VERSION:
+            raise RuntimeError(f"english_section_integrity_seal_version_invalid:{section_title}")
+        expected = str(record.get("sha256") or "")
+        actual = cls._section_integrity_hash(section_text)
+        if not expected or expected != actual:
+            raise RuntimeError(f"english_section_integrity_seal_mismatch:{section_title}")
+
     @staticmethod
     def _replace_keywords(manuscript: str, keywords) -> str:
         value = str(manuscript or '')
@@ -1444,8 +1482,8 @@ class GenerationRunner:
         except PipelineCancelledError:
             raise
         except Exception as error:
-            if not defer_failures:
-                raise
+            # Persist repair state for both intermediate and strict final gates.
+            # The caller owns checkpoint authority and decides whether to stop.
             code_method = getattr(agent, 'translation_error_code', None)
             error_code = str(code_method(error) if callable(code_method) else 'TRANSLATION_PENDING')
             updated_sections[unit_key] = {
@@ -1466,8 +1504,8 @@ class GenerationRunner:
                 'deferred_at': datetime.now(timezone.utc).isoformat(),
             }
             logging.warning(
-                'Translation deferred until final gate | unit=%s code=%s error=%s',
-                unit_name, error_code, type(error).__name__,
+                'Translation repair pending | unit=%s final_gate=%s code=%s error=%s',
+                unit_name, bool(final_gate), error_code, type(error).__name__,
             )
             self._translation_progress_sections = dict(updated_sections)
             self._translation_progress_inflight.discard(str(unit_key))
@@ -1922,6 +1960,13 @@ class GenerationRunner:
             state["english_finalized"] = False
             state["translation_complete"] = False
             state["package_validated"] = False
+        if state.get("english_finalized") and state.get("english_section_seal_version") != ENGLISH_SECTION_SEAL_VERSION:
+            # R2 compatibility: upgrade old completed English sections locally.
+            # No writing-model call, new task, or new quota reservation is allowed.
+            logging.warning("English Section Integrity Seal contract advanced; re-validating frozen completed sections locally.")
+            state["english_finalized"] = False
+            state["translation_complete"] = False
+            state["package_validated"] = False
 
         user_lit_context = str(state.get("user_lit_context") or "")
         user_data_md = str(state.get("user_data_md") or "")
@@ -1961,6 +2006,20 @@ class GenerationRunner:
             )
             self._checkpoint("evidence_ready", state, 6)
 
+        if mode == "amj":
+            current_quant_ledger = dict(data_metadata.get("quantitative_evidence_ledger") or {})
+            expected_quant_version = str(getattr(quant_engine, "QUANTITATIVE_EVIDENCE_LEDGER_VERSION", ""))
+            if current_quant_ledger.get("version") != expected_quant_version:
+                data_metadata["quantitative_evidence_ledger"] = quant_engine.build_quantitative_evidence_ledger(
+                    data_metadata, user_data_md
+                )
+                state["data_metadata"] = data_metadata
+                logging.warning(
+                    "Quantitative Evidence Ledger contract advanced; rebuilt locally version=%s",
+                    expected_quant_version,
+                )
+                self._checkpoint("quantitative_evidence_ledger_upgraded", state, 6)
+
         self._qualitative_quote_sources = (
             self._qualitative_quote_source_map(data_metadata) if mode == "asq" else {}
         )
@@ -2227,6 +2286,8 @@ class GenerationRunner:
                     "citation_unit_section_map": citation_unit_section_map,
                     "citation_validated_tags_by_unit": {},
                     "citation_required_claim_counts_by_unit": {},
+                    "section_integrity_seals": {},
+                    "english_section_seal_version": ENGLISH_SECTION_SEAL_VERSION,
                     "research_logic_contract": research_logic_contract,
                     "working_abstract": working_abstract,
                     "working_keywords": working_keywords,
@@ -2271,6 +2332,11 @@ class GenerationRunner:
             str(name): list(tags or [])
             for name, tags in dict(state.get("citation_validated_tags_by_unit") or {}).items()
         }
+        section_integrity_seals = {
+            str(key): dict(record or {})
+            for key, record in dict(state.get("section_integrity_seals") or {}).items()
+            if isinstance(record, dict)
+        }
         if not citation_minimums:
             citation_minimums, citation_reserve_targets = self._adaptive_citation_requirements(
                 adaptive_structure, mode
@@ -2317,6 +2383,77 @@ class GenerationRunner:
         if self._preview_target_text == '':
             self._preview_target_text = str(state.get('full_draft_translated') or '')
 
+        # P0-R1.1-R2 compatibility migration: old checkpoints predate Section
+        # Integrity Seals. Reconcile only already-completed English sections with
+        # deterministic local contracts, seal them, and invalidate target cache
+        # only when the exact English section changes. Never call the writer.
+        seal_migration = False
+        seal_text_changed = False
+        for section_key, section_title in self._unique_top_level_sections(adaptive_structure):
+            if not self._section_is_complete(adaptive_structure, completed_sections, section_key):
+                continue
+            section_text = self._extract_top_level_section(full_draft_en, section_title)
+            if not section_text:
+                continue
+            existing_seal = dict(section_integrity_seals.get(section_key) or {})
+            if (
+                existing_seal.get("version") == ENGLISH_SECTION_SEAL_VERSION
+                and existing_seal.get("sha256") == self._section_integrity_hash(section_text)
+            ):
+                continue
+            migrated_text, enum_rows = agent.repair_and_validate_enumeration_contract(
+                section_text, section_name=f"Resume sealed section: {section_title}"
+            )
+            migrated_text, fact_rows = self._repair_validate_study_fact_contract(
+                mode, agent, quant_engine, migrated_text, data_metadata,
+                section_name=f"Resume sealed section: {section_title}",
+            )
+            if migrated_text.strip() != section_text.strip():
+                full_draft_en = self._replace_top_level_section(
+                    full_draft_en, section_title, migrated_text
+                )
+                translation_sections.pop(section_key, None)
+                seal_text_changed = True
+                quality_diagnostics.append({
+                    "scope":"english_section_seal_migration", "name":section_title,
+                    "status":"deterministic_repair", "count":len(enum_rows) + len(fact_rows),
+                    "examples":(list(enum_rows or []) + list(fact_rows or []))[:3],
+                })
+                section_text = self._extract_top_level_section(full_draft_en, section_title)
+            section_integrity_seals[section_key] = self._build_section_integrity_seal(
+                section_key, section_title, section_text
+            )
+            seal_migration = True
+        if seal_migration:
+            # Rebuild mutable manuscript memory only from sealed English text so
+            # rejected/raw claims cannot propagate into later sections.
+            for memory_key in ("hypotheses", "variables", "model_summary", "method_summary"):
+                agent.global_memory[memory_key] = ""
+            last_sealed = ""
+            for section_key, section_title in self._unique_top_level_sections(adaptive_structure):
+                if not self._section_is_complete(adaptive_structure, completed_sections, section_key):
+                    continue
+                sealed_text = self._extract_top_level_section(full_draft_en, section_title)
+                if sealed_text:
+                    agent._extract_memory_from_text(sealed_text, section_title)
+                    last_sealed = sealed_text
+            if last_sealed:
+                context_tail = " ".join(last_sealed.split()[-200:])
+            state.update({
+                "full_draft_en": full_draft_en,
+                "translation_sections": translation_sections,
+                "section_integrity_seals": section_integrity_seals,
+                "english_section_seal_version": ENGLISH_SECTION_SEAL_VERSION,
+                "context_tail": context_tail,
+                "quality_diagnostics": quality_diagnostics,
+                "agent_state": agent.export_state(),
+            })
+            if seal_text_changed:
+                state["full_draft_translated"] = self._assemble_partial_translation(
+                    translation_sections, adaptive_structure, include_front=True
+                )
+            self._checkpoint("english_section_seal_migrated", state, 15)
+
         # Target-language work remains Executor-owned but no longer blocks the
         # English writer.  A dedicated single translation agent/worker keeps
         # glossary order deterministic and avoids sharing mutable writer state.
@@ -2769,19 +2906,6 @@ Output only final academic prose. Do not report the word count."""
                         en_resolved = recovered
                     used_after = set(agent.used_citations)
                     survived = used_after - used_before
-            if not is_quarantined and section_id in self._citation_eligible_section_ids(mode):
-                citation_required_claim_counts_by_unit[display_name] = agent.count_external_literature_claims(
-                    en_resolved, section_name=display_name
-                )
-                if assigned_tag_list:
-                    citation_validated_tags_by_unit[display_name] = sorted(
-                        survived, key=agent._citation_tag_number
-                    )
-                logging.info(
-                    "Top-level citation accumulation: unit=%s section=%s validated=%s external_claims=%s",
-                    display_name, citation_unit_section_map.get(display_name, section_id),
-                    len(survived), citation_required_claim_counts_by_unit[display_name],
-                )
             if mode == "asq" and section_id == "findings":
                 en_resolved = self._render_planned_qualitative_quotes(
                     en_resolved,
@@ -2789,6 +2913,43 @@ Output only final academic prose. Do not report the word count."""
                     data_metadata=data_metadata,
                     quote_invocation_plan=quote_invocation_plan,
                 )
+
+            # P0-R1.1-R2: every mutating deterministic operation must complete
+            # before the Evidence Fact contract seals the English unit.
+            en_resolved, enumeration_rows = agent.repair_and_validate_enumeration_contract(
+                en_resolved, section_name=display_name
+            )
+            if enumeration_rows:
+                logging.warning(
+                    "Structural enumeration repaired | unit=%s count=%s details=%s",
+                    display_name, len(enumeration_rows), enumeration_rows[:3],
+                )
+                quality_diagnostics.append({
+                    "scope":"structural_enumeration", "name":display_name,
+                    "status":"deterministic_repair", "count":len(enumeration_rows),
+                    "examples":enumeration_rows[:3],
+                })
+            en_resolved, duplicate_rows = self._remove_severe_cross_unit_duplicates(en_resolved, full_draft_en)
+            if duplicate_rows:
+                logging.warning("Severe cross-unit redundancy removed | unit=%s count=%s max_similarity=%s", display_name, len(duplicate_rows), max(row["similarity"] for row in duplicate_rows))
+                quality_diagnostics.append({"scope":"manuscript_redundancy","name":display_name,"status":"bounded_duplicate_removed","count":len(duplicate_rows),"examples":duplicate_rows[:3]})
+            en_resolved, sentence_echo_rows = agent.remove_severe_cross_unit_sentence_echoes(
+                en_resolved, full_draft_en
+            )
+            if sentence_echo_rows:
+                logging.warning(
+                    "Severe cross-unit sentence echo removed | unit=%s count=%s max_similarity=%s",
+                    display_name, len(sentence_echo_rows),
+                    max(row["similarity"] for row in sentence_echo_rows),
+                )
+                quality_diagnostics.append({
+                    "scope":"manuscript_sentence_echo", "name":display_name,
+                    "status":"bounded_duplicate_removed", "count":len(sentence_echo_rows),
+                    "examples":sentence_echo_rows[:3],
+                })
+
+            # Evidence Fact repair/validation is intentionally LAST. Once this
+            # passes, no later operation may mutate this English unit.
             if mode == "asq":
                 en_resolved, study_fact_rows = agent.repair_qualitative_study_fact_contract(
                     en_resolved, data_metadata, section_name=display_name
@@ -2825,37 +2986,28 @@ Output only final academic prose. Do not report the word count."""
                 quant_engine.validate_quantitative_study_fact_contract(
                     en_resolved, data_metadata, section_name=display_name
                 )
-            en_resolved, enumeration_rows = agent.repair_and_validate_enumeration_contract(
-                en_resolved, section_name=display_name
-            )
-            if enumeration_rows:
-                logging.warning(
-                    "Structural enumeration repaired | unit=%s count=%s details=%s",
-                    display_name, len(enumeration_rows), enumeration_rows[:3],
+
+            # Recompute citation survival/claim demand from the exact text that
+            # will be sealed, not from an earlier pre-cleanup intermediate.
+            if not is_quarantined and section_id in self._citation_eligible_section_ids(mode):
+                final_survived = set()
+                for tag in set(survived or set()):
+                    marker = str((agent.citation_map.get(tag) or {}).get("in_text") or "").strip()
+                    if marker and marker in en_resolved:
+                        final_survived.add(tag)
+                survived = final_survived
+                citation_required_claim_counts_by_unit[display_name] = agent.count_external_literature_claims(
+                    en_resolved, section_name=display_name
                 )
-                quality_diagnostics.append({
-                    "scope":"structural_enumeration", "name":display_name,
-                    "status":"deterministic_repair", "count":len(enumeration_rows),
-                    "examples":enumeration_rows[:3],
-                })
-            en_resolved, duplicate_rows = self._remove_severe_cross_unit_duplicates(en_resolved, full_draft_en)
-            if duplicate_rows:
-                logging.warning("Severe cross-unit redundancy removed | unit=%s count=%s max_similarity=%s", display_name, len(duplicate_rows), max(row["similarity"] for row in duplicate_rows))
-                quality_diagnostics.append({"scope":"manuscript_redundancy","name":display_name,"status":"bounded_duplicate_removed","count":len(duplicate_rows),"examples":duplicate_rows[:3]})
-            en_resolved, sentence_echo_rows = agent.remove_severe_cross_unit_sentence_echoes(
-                en_resolved, full_draft_en
-            )
-            if sentence_echo_rows:
-                logging.warning(
-                    "Severe cross-unit sentence echo removed | unit=%s count=%s max_similarity=%s",
-                    display_name, len(sentence_echo_rows),
-                    max(row["similarity"] for row in sentence_echo_rows),
+                if assigned_tag_list:
+                    citation_validated_tags_by_unit[display_name] = sorted(
+                        survived, key=agent._citation_tag_number
+                    )
+                logging.info(
+                    "Top-level citation accumulation: unit=%s section=%s validated=%s external_claims=%s",
+                    display_name, citation_unit_section_map.get(display_name, section_id),
+                    len(survived), citation_required_claim_counts_by_unit[display_name],
                 )
-                quality_diagnostics.append({
-                    "scope":"manuscript_sentence_echo", "name":display_name,
-                    "status":"bounded_duplicate_removed", "count":len(sentence_echo_rows),
-                    "examples":sentence_echo_rows[:3],
-                })
             full_draft_en += en_resolved + "\n\n"
             if assigned_tag_list:
                 logging.info(
@@ -2863,8 +3015,9 @@ Output only final academic prose. Do not report the word count."""
                     display_name, len(assigned_tag_list), len(survived),
                 )
             self._publish_preview(full_draft_en, display_name, force=True)
-            agent._extract_memory_from_text(en_raw, sec_title)
-            context_tail = " ".join(en_raw.split()[-200:])
+            # Only sealed/reconciled prose may feed subsequent manuscript memory.
+            agent._extract_memory_from_text(en_resolved, sec_title)
+            context_tail = " ".join(en_resolved.split()[-200:])
             actual_words = self._english_count_for_target(en_resolved)
             if not bool(sec.get("exclude_from_word_budget")):
                 accumulated_deviation += actual_words - adjusted_target
@@ -2875,6 +3028,11 @@ Output only final academic prose. Do not report the word count."""
             section_key = str(sec.get('section_id') or sec_title)
             if self._section_is_complete(adaptive_structure, completed_sections, section_key):
                 frozen_section = self._extract_top_level_section(full_draft_en, sec_title)
+                if not frozen_section:
+                    raise RuntimeError(f"english_section_seal_source_missing:{sec_title}")
+                section_integrity_seals[section_key] = self._build_section_integrity_seal(
+                    section_key, sec_title, frozen_section
+                )
                 self._enqueue_translation_unit(
                     translation_agent,
                     unit_key=section_key, unit_name=sec_title, source_text=frozen_section,
@@ -2905,6 +3063,8 @@ Output only final academic prose. Do not report the word count."""
                     "citation_unit_section_map": citation_unit_section_map,
                     "citation_required_claim_counts_by_unit": citation_required_claim_counts_by_unit,
                     "citation_validated_tags_by_unit": citation_validated_tags_by_unit,
+                    "section_integrity_seals": section_integrity_seals,
+                    "english_section_seal_version": ENGLISH_SECTION_SEAL_VERSION,
                     "agent_state": agent.export_state(),
                     "data_metadata": data_metadata,
                 }
@@ -2920,30 +3080,35 @@ Output only final academic prose. Do not report the word count."""
             section_order_for_fact = self._unique_top_level_sections(adaptive_structure)
             if not section_order_for_fact:
                 raise RuntimeError("final_study_fact_section_plan_missing")
-            final_body_for_fact = self._extract_body_from_first_section(
-                full_draft_en, section_order_for_fact[0][1]
-            )
-            if mode == "asq":
-                fact_checked, fact_rows = agent.repair_qualitative_study_fact_contract(
-                    final_body_for_fact, data_metadata, section_name="Final English body"
-                )
-                if fact_checked.strip() != final_body_for_fact.strip() or fact_rows:
-                    raise RuntimeError(
-                        "final_qualitative_study_fact_contract_requires_section_regeneration"
+            # Final English validation is validation-only. Every mutable repair
+            # was completed before the top-level section received its integrity
+            # seal; finalization must never repair the whole manuscript again.
+            for section_key, section_title in section_order_for_fact:
+                sealed_section = self._extract_top_level_section(full_draft_en, section_title)
+                if not sealed_section:
+                    raise RuntimeError(f"final_english_section_missing:{section_title}")
+                seal = dict(section_integrity_seals.get(section_key) or {})
+                self._validate_section_integrity_seal(seal, sealed_section, section_title)
+                try:
+                    if mode == "asq":
+                        agent.validate_qualitative_study_fact_contract(
+                            sealed_section, data_metadata, section_name=section_title
+                        )
+                    elif mode == "amj":
+                        quant_engine.validate_quantitative_study_fact_contract(
+                            sealed_section, data_metadata, section_name=section_title
+                        )
+                except Exception as exc:
+                    prefix = (
+                        "final_qualitative_study_fact_contract_failed"
+                        if mode == "asq" else "final_quantitative_study_fact_contract_failed"
                     )
-            elif mode == "amj":
-                fact_checked, fact_rows = quant_engine.repair_quantitative_study_fact_contract(
-                    final_body_for_fact, data_metadata, section_name="Final English body"
+                    raise RuntimeError(f"{prefix}:{section_title}") from exc
+                enum_checked, _enum_rows = agent.repair_and_validate_enumeration_contract(
+                    sealed_section, section_name=f"Final sealed section: {section_title}"
                 )
-                if fact_checked.strip() != final_body_for_fact.strip() or fact_rows:
-                    raise RuntimeError(
-                        "final_quantitative_study_fact_contract_requires_section_regeneration"
-                    )
-            enum_checked, enum_rows = agent.repair_and_validate_enumeration_contract(
-                full_draft_en, section_name="Final English manuscript"
-            )
-            if enum_checked.strip() != full_draft_en.strip() or enum_rows:
-                raise RuntimeError("final_enumeration_contract_requires_section_regeneration")
+                if enum_checked.strip() != sealed_section.strip():
+                    raise RuntimeError(f"final_enumeration_contract_failed:{section_title}")
             # Validate the completed body while the Working Abstract is still in
             # place. Final front matter must not be generated until all body
             # sections, citation cleanup, and any Discussion rebalance are frozen.
@@ -3127,6 +3292,8 @@ Output only final academic prose. Do not report the word count."""
                     "citation_unit_section_map": citation_unit_section_map,
                     "citation_required_claim_counts_by_unit": citation_required_claim_counts_by_unit,
                     "citation_validated_tags_by_unit": citation_validated_tags_by_unit,
+                    "section_integrity_seals": section_integrity_seals,
+                    "english_section_seal_version": ENGLISH_SECTION_SEAL_VERSION,
                     "quality_diagnostics": quality_diagnostics,
                     "english_finalized": True,
                     "translation_complete": False,
@@ -3178,6 +3345,26 @@ Output only final academic prose. Do not report the word count."""
                     'agent_state': agent.export_state(),
                 })
                 pct = min(96, 94 + int(index * 2 / max(1, len(final_units))))
+                current_translation = dict(translation_sections.get(unit_key) or {})
+                blocking = (
+                    current_translation.get('qa_status') not in {'passed', 'passed_with_warning'}
+                    or not current_translation.get('translated_text')
+                )
+                if blocking:
+                    state['translation_blocking_unit'] = str(unit_key)
+                    state['translation_blocking_unit_name'] = str(unit_name)
+                    state['translation_blocking_error_code'] = str(current_translation.get('error_code') or 'TRANSLATION_PENDING')
+                    self._checkpoint(
+                        f'translation_{unit_key}_repair_pending', state, pct,
+                        display_node=f'translation_repair_pending:{index}:{len(final_units)}',
+                    )
+                    # Stop at the first blocking unit. The durable claim state is
+                    # now checkpointed, so resume continues exactly here instead
+                    # of spending tokens on later units or repeating prior repair.
+                    raise RuntimeError('translation_final_gate_unresolved:' + str(unit_key))
+                state.pop('translation_blocking_unit', None)
+                state.pop('translation_blocking_unit_name', None)
+                state.pop('translation_blocking_error_code', None)
                 self._checkpoint(
                     f'translation_{unit_key}', state, pct,
                     display_node=f'translation_final_gate:{index}:{len(final_units)}',

core/pipeline_agent.py

@@ -3430,6 +3430,13 @@ INPUT:
 
     @classmethod
     def _nearest_qual_source_family(cls, sentence, actor_start):
+        """Resolve the data source grammatically associated with one actor.
+
+        Prefer the last source phrase before the actor (``interviews with X``),
+        then fall back to the nearest following source (``X was interviewed``).
+        Absolute-distance matching can cross an ``and`` boundary and wrongly
+        pair X with the next actor's source.
+        """
         text = str(sentence or "")
         markers = []
         patterns = {
@@ -3439,22 +3446,75 @@ INPUT:
             "log": r"\b(?:usage|interaction|system)\s+logs?\b|\blog\s+data\b",
             "document": r"\bdocuments?\b|\barchival\s+(?:records?|materials?)\b|\brecords?\b",
         }
+        position = int(actor_start)
         for family, pattern in patterns.items():
             for match in re.finditer(pattern, text, flags=re.I):
-                markers.append((abs(match.start() - int(actor_start)), family))
+                markers.append((match.start(), match.end(), family))
         if not markers:
             return ""
-        markers.sort(key=lambda row: row[0])
-        return markers[0][1]
+        preceding = [row for row in markers if row[1] <= position]
+        if preceding:
+            start, end, family = max(preceding, key=lambda row: row[1])
+            # A source more than one short clause away is unlikely to own actor.
+            if position - end <= 120:
+                return family
+        following = [row for row in markers if row[0] >= position]
+        if following:
+            return min(following, key=lambda row: row[0] - position)[2]
+        return ""
+
+    @staticmethod
+    def _has_external_literature_marker(sentence):
+        value = str(sentence or "")
+        if re.search(r"\bCITE-\d+\b", value, flags=re.I):
+            return True
+        if re.search(
+            r"\((?=[^()\n]*[A-Za-zÀ-ÖØ-öø-ÿ'’.-]+(?:\s+(?:&\s+[A-Za-zÀ-ÖØ-öø-ÿ'’.-]+|et\s+al\.))?\s*,\s*(?:19|20)\d{2}[a-z]?\b)[^()\n]+\)",
+            value, flags=re.I,
+        ):
+            return True
+        if re.search(r"\b[A-Z][A-Za-z'’.-]+\s+et\s+al\.\s*\((?:19|20)\d{2}[a-z]?\)", value):
+            return True
+        return False
+
+    @classmethod
+    def _is_study_owned_qualitative_claim(cls, sentence, section_name=""):
+        """Conservatively identify only this manuscript's empirical/procedural claims.
+
+        External literature, quoted participant speech, and generic methodological
+        statements are immutable to the Study Evidence Ledger. False negatives are
+        safer here than rewriting another study's facts.
+        """
+        value = str(sentence or "")
+        stripped = value.strip()
+        if not stripped or stripped.startswith(">"):
+            return False
+        if re.match(r'^["“‘\']', stripped) or re.search(r"\bQUOTE-\d+\b", value, flags=re.I):
+            return False
+        if cls._has_external_literature_marker(value):
+            return False
+        if re.search(r"(?i)\b(?:prior|previous|earlier|existing)\s+(?:research|stud(?:y|ies)|literature)\b", value):
+            return False
+        strong = re.compile(
+            r"(?i)\b(?:we|our\s+study|this\s+study|the\s+present\s+study|the\s+study|"
+            r"research\s+team|our\s+data|our\s+analysis|our\s+interviews?|our\s+focus\s+groups?|"
+            r"our\s+observations?|data\s+collection|the\s+empirical\s+base|the\s+sample|"
+            r"(?:interview|focus\s+group|observation)\s+participants?|"
+            r"participants?\s+(?:in\s+this\s+study\s+)?(?:were|took|participated)|"
+            r"respondents?\s+(?:in\s+this\s+study\s+)?(?:were|took|participated)|"
+            r"interviews?\s+(?:were|was)\s+conducted|focus\s+groups?\s+(?:were|was)\s+conducted|"
+            r"observations?\s+(?:were|was)\s+(?:conducted|recorded))\b|"
+            r"\b(?:was|were)\s+(?:conducted|performed|used|maintained|obtained|recorded|anonymized|anonymised)\b"
+        )
+        return bool(strong.search(value))
 
     @classmethod
     def repair_qualitative_study_fact_contract(cls, text, metadata, section_name=""):
-        """Remove/repair only high-confidence study-owned fact violations.
+        """Repair only high-confidence *study-owned* qualitative fact violations.
 
-        The frozen Study Evidence Ledger owns actor/data-source relations and
-        whether trustworthiness/ethics/reflexivity procedures were actually
-        performed. External methodological best practice never authorizes a
-        study-owned procedure claim.
+        The frozen Study Evidence Ledger owns this study's actor/data-source
+        relations and explicitly performed procedures. External literature and
+        participant quotations are never rewritten by this deterministic layer.
         """
         value = str(text or "")
         facts = dict((metadata or {}).get("study_facts") or {})
@@ -3466,6 +3526,7 @@ INPUT:
         }
         allowed_by_actor = {}
         canonical_relations = []
+        canonical_by_actor = {}
         for row in relations:
             role = re.sub(r"\s+", " ", str(row.get("actor_role") or "")).strip()
             source = re.sub(r"\s+", " ", str(row.get("data_source") or "")).strip()
@@ -3475,80 +3536,142 @@ INPUT:
             canonical_relations.append((role, source, row.get("count")))
             for variant in cls._qual_actor_variants(role):
                 allowed_by_actor.setdefault(variant, set()).add(family)
+                canonical_by_actor.setdefault(variant, []).append((role, source, family, row.get("count")))
 
-        study_owned = re.compile(
-            r"\b(?:we|our\s+study|this\s+study|the\s+study|research\s+team|participants?|respondents?)\b|"
-            r"\b(?:was|were)\s+(?:conducted|performed|used|maintained|obtained|recorded|anonymized|anonymised)\b",
-            flags=re.I,
-        )
         procedure_patterns = {
             "trustworthiness": r"\bmember\s+check(?:ing|s)?\b|\bpeer\s+debrief(?:ing|ed)?\b|\baudit\s+trail\b|\bexternal\s+audit\b",
             "ethics": r"\binformed\s+consent\b|\bethics?\s+approval\b|\binstitutional\s+review\s+board\b|\bIRB\b|\bapproval\s+(?:number|code)\b",
             "reflexivity": r"\bfield\s+journal\b|\breflexive\s+journal\b|\bpositionality\s+(?:statement|memo)\b|\breflexive\s+memo(?:s)?\b",
         }
+        source_patterns = {
+            "interview": r"\b(?:semi[- ]structured\s+)?interviews?\b|\binterviewed\b|\binterview\s+participants?\b",
+            "focus_group": r"\bfocus\s+groups?\b",
+            "observation": r"\b(?:field\s+)?observations?\b|\bobserved\b",
+            "log": r"\b(?:usage|interaction|system)\s+logs?\b|\blog\s+data\b",
+            "document": r"\bdocuments?\b|\barchival\s+(?:records?|materials?)\b|\brecords?\b",
+        }
         spans = cls._sentence_spans(value)
         replacements = []
         diagnostics = []
         for start, end, sentence in spans:
             core = str(sentence or "")
             lower = core.casefold()
-            if not core.strip():
+            if not core.strip() or not cls._is_study_owned_qualitative_claim(core, section_name):
                 continue
             removed = False
-            if study_owned.search(core):
-                for category, pattern in procedure_patterns.items():
-                    procedure_match = re.search(pattern, core, flags=re.I)
-                    if not procedure_match:
-                        continue
-                    observed = procedure_match.group(0).casefold()
-                    if "member check" in observed:
-                        key = "member check"
-                    elif "peer debrief" in observed:
-                        key = "peer debrief"
-                    elif "audit trail" in observed:
-                        key = "audit trail"
-                    elif "external audit" in observed:
-                        key = "external audit"
-                    elif "informed consent" in observed:
-                        key = "informed consent"
-                    elif "ethic" in observed and "approval" in observed:
-                        key = "ethics approval"
-                    elif "review board" in observed or observed == "irb":
-                        key = "review board"
-                    elif "approval" in observed:
-                        key = "approval"
-                    elif "field journal" in observed:
-                        key = "field journal"
-                    elif "reflexive journal" in observed:
-                        key = "reflexive journal"
-                    elif "positionality" in observed:
-                        key = "positionality"
-                    else:
-                        key = "reflexive memo"
-                    authorized = any(key in item.casefold() for item in procedure_lists[category])
-                    if not authorized:
-                        replacements.append((start, end, ""))
-                        diagnostics.append({"type":"unsupported_study_procedure", "category":category, "procedure":key, "preview":core.strip()[:180]})
-                        removed = True
-                        break
+            for category, pattern in procedure_patterns.items():
+                procedure_match = re.search(pattern, core, flags=re.I)
+                if not procedure_match:
+                    continue
+                observed = procedure_match.group(0).casefold()
+                if "member check" in observed:
+                    key = "member check"
+                elif "peer debrief" in observed:
+                    key = "peer debrief"
+                elif "audit trail" in observed:
+                    key = "audit trail"
+                elif "external audit" in observed:
+                    key = "external audit"
+                elif "informed consent" in observed:
+                    key = "informed consent"
+                elif "ethic" in observed and "approval" in observed:
+                    key = "ethics approval"
+                elif "review board" in observed or observed == "irb":
+                    key = "review board"
+                elif "approval" in observed:
+                    key = "approval"
+                elif "field journal" in observed:
+                    key = "field journal"
+                elif "reflexive journal" in observed:
+                    key = "reflexive journal"
+                elif "positionality" in observed:
+                    key = "positionality"
+                else:
+                    key = "reflexive memo"
+                authorized = any(key in item.casefold() for item in procedure_lists[category])
+                if not authorized:
+                    replacements.append((start, end, ""))
+                    diagnostics.append({"type":"unsupported_study_procedure", "category":category, "procedure":key, "preview":core.strip()[:180]})
+                    removed = True
+                    break
             if removed or not canonical_relations:
                 continue
+
+            # A pure participant-membership sentence can conflate two actor
+            # groups under one source (e.g., "interview participants included
+            # officers and farmers"). In that narrow case, render the frozen
+            # relation set directly; this is safer than a phrase substitution
+            # that would leave a grammatically or factually mixed sentence.
+            participant_membership = bool(re.search(
+                r"(?i)\b(?:interview|focus\s+group|observation)\s+participants?\b|"
+                r"\bparticipants?\s+(?:included|comprised|consisted)\b", core
+            ))
+            present_roles = []
+            for role, source, count in canonical_relations:
+                if any(re.search(rf"(?<![A-Za-z]){re.escape(variant)}(?![A-Za-z])", lower, flags=re.I)
+                       for variant in cls._qual_actor_variants(role)):
+                    present_roles.append((role, source, count))
+            if participant_membership and len(present_roles) >= 2:
+                clauses = [
+                    f"{source} with {role}" + (f" (n={count})" if isinstance(count, int) else "")
+                    for role, source, count in present_roles
+                ]
+                canonical_sentence = "Data collection used " + " and ".join(clauses) + "."
+                leading = core[: len(core) - len(core.lstrip())]
+                trailing = core[len(core.rstrip()):]
+                replacement = f"{leading}{canonical_sentence}{trailing}"
+                if replacement != core:
+                    replacements.append((start, end, replacement))
+                    diagnostics.append({
+                        "type":"actor_data_source_relation_repaired",
+                        "actor":"multiple", "observed":"mixed_participant_source",
+                        "allowed":[cls._qual_source_family(source) for _role, source, _count in present_roles],
+                    })
+                continue
+
+            # Repair only the mismatched source phrase nearest to the actor. This
+            # preserves the sentence's substantive argument instead of replacing
+            # the entire sentence with a mechanical data-collection template.
             for actor_variant, allowed_families in allowed_by_actor.items():
-                for actor_match in re.finditer(rf"(?<![A-Za-z]){re.escape(actor_variant)}(?![A-Za-z])", lower, flags=re.I):
-                    nearest = cls._nearest_qual_source_family(core, actor_match.start())
-                    if nearest and nearest not in allowed_families:
-                        relation_text = " and ".join(
-                            f"{source} with {role}" + (f" (n={count})" if isinstance(count, int) else "")
-                            for role, source, count in canonical_relations
-                        )
-                        replacement = f"Data collection comprised {relation_text}."
-                        leading = core[: len(core) - len(core.lstrip())]
-                        trailing = core[len(core.rstrip()):]
-                        replacements.append((start, end, f"{leading}{replacement}{trailing}"))
-                        diagnostics.append({"type":"actor_data_source_relation_repaired", "actor":actor_variant, "observed":nearest, "allowed":sorted(allowed_families)})
-                        removed = True
-                        break
-                if removed:
+                actor_matches = list(re.finditer(rf"(?<![A-Za-z]){re.escape(actor_variant)}(?![A-Za-z])", lower, flags=re.I))
+                if not actor_matches:
+                    continue
+                changed_core = core
+                actor_changed = False
+                for actor_match in actor_matches:
+                    nearest = cls._nearest_qual_source_family(changed_core, actor_match.start())
+                    if not nearest or nearest in allowed_families:
+                        continue
+                    preferred_rows = canonical_by_actor.get(actor_variant) or []
+                    preferred = next((row for row in preferred_rows if row[2] in allowed_families), None)
+                    if not preferred:
+                        continue
+                    canonical_source = preferred[1]
+                    pattern = source_patterns.get(nearest)
+                    if not pattern:
+                        continue
+                    source_matches = list(re.finditer(pattern, changed_core, flags=re.I))
+                    if not source_matches:
+                        continue
+                    source_match = min(source_matches, key=lambda m: abs(m.start() - actor_match.start()))
+                    observed_phrase = source_match.group(0)
+                    replacement_source = canonical_source
+                    if re.fullmatch(r"(?i)interviewed", observed_phrase.strip()):
+                        replacement_source = f"conducted {canonical_source} with"
+                    elif re.fullmatch(r"(?i)observed", observed_phrase.strip()):
+                        replacement_source = f"used {canonical_source} with"
+                    candidate = changed_core[:source_match.start()] + replacement_source + changed_core[source_match.end():]
+                    if candidate != changed_core:
+                        diagnostics.append({
+                            "type":"actor_data_source_relation_repaired", "actor":actor_variant,
+                            "observed":nearest, "allowed":sorted(allowed_families),
+                            "replacement":canonical_source,
+                        })
+                        changed_core = candidate
+                        actor_changed = True
+                    break
+                if actor_changed and changed_core != core:
+                    replacements.append((start, end, changed_core))
                     break
         if replacements:
             for start, end, replacement in sorted(replacements, reverse=True):

core/quantitative_engine.py

@@ -17,7 +17,7 @@ from utils.file_parser import extract_data_from_files
 class QuantitativeEngine:
     """Prepare one frozen quantitative evidence packet before manuscript writing."""
 
-    QUANTITATIVE_EVIDENCE_LEDGER_VERSION = "quantitative-evidence-ledger-v1"
+    QUANTITATIVE_EVIDENCE_LEDGER_VERSION = "quantitative-evidence-ledger-v2"
 
     @staticmethod
     def generate_psd_corr_matrix(n_constructs, max_corr=0.5):
@@ -344,14 +344,70 @@ Keywords: {keywords}"""
             spans.append((start, len(value), value[start:]))
         return spans
 
+    @staticmethod
+    def _has_external_literature_marker(sentence):
+        value = str(sentence or "")
+        if re.search(r"\bCITE-\d+\b", value, flags=re.I):
+            return True
+        if re.search(
+            r"\((?=[^()\n]*[A-Za-zÀ-ÖØ-öø-ÿ'’.-]+(?:\s+(?:&\s+[A-Za-zÀ-ÖØ-öø-ÿ'’.-]+|et\s+al\.))?\s*,\s*(?:19|20)\d{2}[a-z]?\b)[^()\n]+\)",
+            value, flags=re.I,
+        ):
+            return True
+        if re.search(r"\b[A-Z][A-Za-z'’.-]+\s+et\s+al\.\s*\((?:19|20)\d{2}[a-z]?\)", value):
+            return True
+        return False
+
+    @classmethod
+    def _quantitative_sentence_ownership(cls, sentence, section_name="", known_constructs=None):
+        """Return external/study/result/unknown for deterministic AMJ repair.
+
+        Only study/result-owned prose may be rewritten. External literature is
+        immutable even when it reports the same constructs or statistics.
+        """
+        value = str(sentence or "")
+        stripped = value.strip()
+        if not stripped or stripped.startswith(">"):
+            return "external" if stripped else "unknown"
+        if cls._has_external_literature_marker(value):
+            return "external"
+        if re.search(r"(?i)\b(?:prior|previous|earlier|existing)\s+(?:research|stud(?:y|ies)|literature)\b", value):
+            return "external"
+        if re.search(r"(?i)\b(?:we|our\s+study|this\s+study|the\s+present\s+study|our\s+sample|our\s+analysis|our\s+results|the\s+sample|the\s+survey|respondents?|participants?)\b", value):
+            return "study"
+        section = str(section_name or "").casefold()
+        constructs = [str(x or "").casefold() for x in (known_constructs or []) if str(x or "").strip()]
+        has_construct = any(item in value.casefold() for item in constructs)
+        has_metric = bool(re.search(r"(?i)(?:\bVIF\b|\bbootstrap\b|R(?:²|\^2)|(?:β|standardized\s+estimate)\s*(?:=|:)|\bp\s*(?:=|<|>|≤|≥))", value))
+        has_role = bool(re.search(r"(?i)\b(?:predictor|mediator|outcome|independent\s+variable|dependent\s+variable)\b", value))
+        if any(token in section for token in ("result", "analysis", "abstract", "final")) and (has_construct or has_metric or has_role):
+            return "result"
+        if any(token in section for token in ("method", "measure")) and (has_construct or has_role or re.search(r"(?i)\b(?:Likert|sample|survey)\b", value)):
+            return "study"
+        return "unknown"
+
+    @staticmethod
+    def _unique_numeric(pattern, evidence, *, integer=False):
+        values = []
+        for raw in re.findall(pattern, str(evidence or ""), flags=re.I):
+            raw_value = raw[-1] if isinstance(raw, tuple) else raw
+            try:
+                value = int(float(raw_value)) if integer else float(raw_value)
+            except Exception:
+                continue
+            values.append(value)
+        unique = sorted(set(values))
+        return unique[0] if len(unique) == 1 else None
+
     @classmethod
     def build_quantitative_evidence_ledger(cls, metadata, evidence_text=""):
         """Freeze only high-confidence study-owned quantitative facts.
 
         System-generated evidence has typed metadata and therefore supports a
-        strict contract.  User-provided evidence is parsed conservatively: only
-        an unambiguous sample size / Likert scale is promoted to deterministic
-        authority; everything else remains owned by the uploaded evidence text.
+        strict contract. User-provided evidence is parsed conservatively: only
+        uniquely labelled values are promoted to deterministic authority.
+        Ambiguous values remain owned by the uploaded evidence text and are not
+        guessed or globally substituted.
         """
         meta = dict(metadata or {})
         source_type = str(meta.get("source_type") or "")
@@ -359,6 +415,7 @@ Keywords: {keywords}"""
         scale_points = meta.get("scale_points")
         strict = source_type == "system_generated"
         evidence = str(evidence_text or "")
+        uploaded_metric_anchors = {}
         if not strict and evidence:
             sample_candidates = set()
             for pattern in (
@@ -371,6 +428,39 @@ Keywords: {keywords}"""
             scale_candidates = {int(x) for x in re.findall(r"(?i)\b([2-9]|1[0-1])[- ]point\s+Likert", evidence)}
             if len(scale_candidates) == 1:
                 scale_points = next(iter(scale_candidates))
+
+            vif = cls._unique_numeric(r"(?i)\bVIF\s*(?:=|:)?\s*([-+]?\d+(?:\.\d+)?)", evidence)
+            bootstrap = cls._unique_numeric(r"(?i)\bbootstrap\s+(?:resamples?|iterations?)\s*(?:=|:)?\s*(\d{2,6})", evidence, integer=True)
+            if vif is not None:
+                uploaded_metric_anchors["VIF"] = vif
+            if bootstrap is not None:
+                uploaded_metric_anchors["bootstrap_iterations"] = int(bootstrap)
+
+            r2_by_label = {}
+            for match in re.finditer(
+                r"(?i)R(?:²|\^2)\s*(?:for|of)\s*([A-Za-z][A-Za-z0-9 _\-/]{1,80}?)\s*(?:=|:)\s*([-+]?\d+(?:\.\d+)?)",
+                evidence,
+            ):
+                label = re.sub(r"\s+", " ", match.group(1)).strip().casefold()
+                r2_by_label.setdefault(label, set()).add(float(match.group(2)))
+            uploaded_metric_anchors["r2_by_label"] = {
+                label: next(iter(values)) for label, values in r2_by_label.items() if len(values) == 1
+            }
+
+            beta_by_path = {}
+            for match in re.finditer(
+                r"(?i)([A-Za-z][A-Za-z0-9 _\-/]{1,60}?)\s*(?:→|->)\s*([A-Za-z][A-Za-z0-9 _\-/]{1,60}?)\s*[,;:]?\s*(?:β|standardized\s+estimate)\s*(?:=|:)\s*([-+]?\d+(?:\.\d+)?)",
+                evidence,
+            ):
+                key = (
+                    re.sub(r"\s+", " ", match.group(1)).strip().casefold(),
+                    re.sub(r"\s+", " ", match.group(2)).strip().casefold(),
+                )
+                beta_by_path.setdefault(key, set()).add(float(match.group(3)))
+            uploaded_metric_anchors["beta_by_path"] = {
+                " -> ".join(key): next(iter(values)) for key, values in beta_by_path.items() if len(values) == 1
+            }
+
         constructs = dict(meta.get("constructs") or {}) if isinstance(meta.get("constructs"), dict) else {}
         results = dict(meta.get("association_results") or {}) if isinstance(meta.get("association_results"), dict) else {}
         return {
@@ -382,6 +472,7 @@ Keywords: {keywords}"""
             "study_design": str(meta.get("study_design") or "").strip(),
             "constructs": {k: str(v or "").strip() for k, v in constructs.items() if str(v or "").strip()},
             "association_results": results,
+            "uploaded_metric_anchors": uploaded_metric_anchors,
             "causal_identification": "none" if strict else "evidence_owned",
             "allowed_methods": [
                 "reliability", "composite reliability", "average variance extracted",
@@ -398,7 +489,7 @@ Keywords: {keywords}"""
 
     @classmethod
     def repair_quantitative_study_fact_contract(cls, text, metadata, section_name=""):
-        """Repair only deterministic AMJ study-fact drift owned by frozen evidence."""
+        """Repair deterministic AMJ drift only in study/result-owned prose."""
         value = str(text or "")
         meta = dict(metadata or {})
         ledger = dict(meta.get("quantitative_evidence_ledger") or {})
@@ -410,20 +501,34 @@ Keywords: {keywords}"""
         strict = bool(ledger.get("strict"))
         constructs = dict(ledger.get("constructs") or {})
         results = dict(ledger.get("association_results") or {})
+        uploaded = dict(ledger.get("uploaded_metric_anchors") or {})
+        known_constructs = [str(v) for v in constructs.values() if str(v)]
+        construct_folded = [c.casefold() for c in known_constructs]
+        forbidden = [str(x) for x in ledger.get("forbidden_unowned_methods") or []]
+        role_patterns = {
+            "predictor": r"(?:predictor|independent\s+variable|predictor\s+construct)",
+            "mediator": r"(?:mediator|mediating\s+variable|mediating\s+construct)",
+            "outcome": r"(?:outcome|dependent\s+variable|outcome\s+construct)",
+        }
+        causal_replacements = (
+            (r"(?i)\bhad\s+a\s+causal\s+effect\s+on\b", "was associated with"),
+            (r"(?i)\bexerted\s+a\s+causal\s+effect\s+on\b", "was associated with"),
+            (r"(?i)\bcausal\s+effect\s+on\b", "association with"),
+            (r"(?i)\ban\s+causal\s+effect\b", "an association"),
+            (r"(?i)\ba\s+causal\s+effect\b", "an association"),
+            (r"(?i)\bthe\s+causal\s+effect\b", "the association"),
+            (r"(?i)\bcausal\s+effects\b", "associations"),
+            (r"(?i)\bcausal\s+effect\b", "association"),
+            (r"(?i)\bcausal\s+mediation\b", "indirect association"),
+        )
 
-        # Sample/scale claims are repaired only when the sentence is clearly
-        # study-owned and not an external cited finding.
-        spans = cls._sentence_spans(value)
-        sample_scale_replacements = []
-        for start, end, sentence in spans:
-            has_citation = bool(re.search(r"\([^()]*?(?:19|20)\d{2}[a-z]?[^()]*?\)", sentence))
-            study_owned = bool(re.search(
-                r"(?i)\b(?:this\s+study|our\s+study|we|our\s+sample|the\s+sample|sample\s+(?:comprised|consisted|included)|respondents?|participants?)\b",
-                sentence,
-            ))
-            if has_citation or not study_owned:
+        replacements = []
+        for start, end, sentence in cls._sentence_spans(value):
+            ownership = cls._quantitative_sentence_ownership(sentence, section_name, known_constructs)
+            if ownership not in {"study", "result"}:
                 continue
             revised = sentence
+
             if sample_size:
                 sample_pattern = re.compile(r"(?i)(\b(?:sample(?:\s+size)?(?:\s+of)?|N)\s*(?:=|:)?\s*)(\d{2,6})\b")
                 def sample_repl(match):
@@ -450,79 +555,36 @@ Keywords: {keywords}"""
                     diagnostics.append({"type":"scale_points_repaired","observed":observed,"expected":int(scale_points)})
                     return f"{int(scale_points)}-point {match.group(2)}"
                 revised = scale_pattern.sub(scale_repl, revised)
-            if revised != sentence:
-                sample_scale_replacements.append((start, end, revised))
-        for start, end, replacement in sorted(sample_scale_replacements, reverse=True):
-            value = value[:start] + replacement + value[end:]
 
-        # Explicit construct-role labels are deterministic and safe to correct.
-        role_patterns = {
-            "predictor": r"(?:predictor|independent\s+variable|predictor\s+construct)",
-            "mediator": r"(?:mediator|mediating\s+variable|mediating\s+construct)",
-            "outcome": r"(?:outcome|dependent\s+variable|outcome\s+construct)",
-        }
-        known_constructs = [str(v) for v in constructs.values() if str(v)]
-        for role, role_pattern in role_patterns.items():
-            expected = str(constructs.get(role) or "")
-            if not expected:
-                continue
-            for candidate in known_constructs:
-                if candidate.casefold() == expected.casefold():
+            for role, role_pattern in role_patterns.items():
+                expected = str(constructs.get(role) or "")
+                if not expected:
                     continue
-                pattern = re.compile(rf"(?i)({role_pattern}\s*(?:was|is|:|=)?\s*){re.escape(candidate)}")
-                if pattern.search(value):
-                    value, count = pattern.subn(lambda m: m.group(1) + expected, value)
+                for candidate in known_constructs:
+                    if candidate.casefold() == expected.casefold():
+                        continue
+                    pattern = re.compile(rf"(?i)({role_pattern}\s*(?:was|is|:|=)?\s*){re.escape(candidate)}")
+                    revised, count = pattern.subn(lambda m: m.group(1) + expected, revised)
                     if count:
                         diagnostics.append({"type":"construct_role_repaired","role":role,"observed":candidate,"expected":expected,"count":count})
 
-        if strict:
-            # Frozen system-generated evidence has no causal identification.
-            replacements = (
-                (r"(?i)\bhad\s+a\s+causal\s+effect\s+on\b", "was associated with"),
-                (r"(?i)\bexerted\s+a\s+causal\s+effect\s+on\b", "was associated with"),
-                (r"(?i)\bcausal\s+effect\s+on\b", "association with"),
-                (r"(?i)\ban\s+causal\s+effect\b", "an association"),
-                (r"(?i)\ba\s+causal\s+effect\b", "an association"),
-                (r"(?i)\bthe\s+causal\s+effect\b", "the association"),
-                (r"(?i)\bcausal\s+effects\b", "associations"),
-                (r"(?i)\bcausal\s+effect\b", "association"),
-                (r"(?i)\bcausal\s+mediation\b", "indirect association"),
-            )
-            for pattern, replacement in replacements:
-                value, count = re.subn(pattern, replacement, value)
-                if count:
-                    diagnostics.append({"type":"causal_language_repaired","replacement":replacement,"count":count})
-
-            # Remove p-values attached to this study's frozen constructs because
-            # no p-values are generated/frozen by this engine. External cited
-            # literature claims are left untouched.
-            construct_folded = [c.casefold() for c in known_constructs]
-            spans = cls._sentence_spans(value)
-            replacements_by_span = []
-            forbidden = [str(x) for x in ledger.get("forbidden_unowned_methods") or []]
-            section_sensitive = any(token in str(section_name or "").casefold() for token in ("method", "result", "abstract", "final", "analysis"))
-            for start, end, sentence in spans:
-                lower = sentence.casefold()
-                has_construct = any(c in lower for c in construct_folded) if construct_folded else False
-                has_citation = bool(re.search(r"\([^()]*?(?:19|20)\d{2}[a-z]?[^()]*?\)", sentence))
-                revised = sentence
-                if has_construct and not has_citation:
+            if strict:
+                for pattern, replacement in causal_replacements:
+                    revised, count = re.subn(pattern, replacement, revised)
+                    if count:
+                        diagnostics.append({"type":"causal_language_repaired","replacement":replacement,"count":count})
+                if any(c in revised.casefold() for c in construct_folded) if construct_folded else False:
                     revised, p_count = re.subn(r"(?i)\s*[,;]?\s*p\s*(?:=|<|>|≤|≥)\s*(?:0?\.\d+|\d+(?:\.\d+)?)", "", revised)
                     if p_count:
                         diagnostics.append({"type":"unfrozen_p_value_removed","count":p_count})
-                if section_sensitive and not has_citation and re.search(r"(?i)\b(?:we|our\s+study|this\s+study|the\s+study|our\s+analysis|the\s+analysis)\b", sentence):
-                    bad = next((term for term in forbidden if re.search(rf"(?i)(?<![A-Za-z]){re.escape(term)}(?![A-Za-z])", sentence)), "")
+                if ownership == "study":
+                    bad = next((term for term in forbidden if re.search(rf"(?i)(?<![A-Za-z]){re.escape(term)}(?![A-Za-z])", revised)), "")
                     if bad:
-                        revised = ""
                         diagnostics.append({"type":"unsupported_quantitative_method_removed","method":bad,"preview":sentence.strip()[:180]})
-                if revised != sentence:
-                    replacements_by_span.append((start, end, revised))
-            for start, end, replacement in sorted(replacements_by_span, reverse=True):
-                value = value[:start] + replacement + value[end:]
-
-            # High-confidence frozen numeric anchors.
-            def replace_metric(pattern, expected, dtype="float", label="metric"):
-                nonlocal value
+                        revised = ""
+
+            def replace_local_metric(pattern, expected, *, dtype="float", label="metric"):
+                nonlocal revised
                 if expected is None:
                     return
                 expected_text = str(int(expected)) if dtype == "int" else f"{float(expected):.3f}".rstrip("0").rstrip(".")
@@ -537,61 +599,56 @@ Keywords: {keywords}"""
                         return match.group(0)
                     diagnostics.append({"type":"quantitative_metric_repaired","metric":label,"observed":observed,"expected":expected_text})
                     return match.group(1) + expected_text
-                value = regex.sub(repl, value)
+                revised = regex.sub(repl, revised)
+
+            vif_expected = results.get("predictor_mediator_vif") if strict else uploaded.get("VIF")
+            boot_expected = results.get("bootstrap_iterations") if strict else uploaded.get("bootstrap_iterations")
+            replace_local_metric(r"(\bVIF\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", vif_expected, label="VIF")
+            replace_local_metric(r"(\bbootstrap\s+(?:resamples?|iterations?)\s*(?:=|:)?\s*)(\d{2,6})", boot_expected, dtype="int", label="bootstrap_iterations")
 
-            if results:
-                replace_metric(r"(\bVIF\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", results.get("predictor_mediator_vif"), label="VIF")
-                replace_metric(r"(\bbootstrap\s+(?:resamples?|iterations?)\s*(?:=|:)?\s*)(\d{2,6})", results.get("bootstrap_iterations"), dtype="int", label="bootstrap_iterations")
+            if strict and results:
                 for role, key in (("mediator", "r2_mediator"), ("outcome", "r2_outcome")):
                     construct = str(constructs.get(role) or "")
                     if construct and results.get(key) is not None:
                         metric_pattern = rf"(R(?:²|\^2)\s*(?:for|of)?\s*{re.escape(construct)}\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)"
-                        replace_metric(metric_pattern, results.get(key), label=key)
+                        replace_local_metric(metric_pattern, results.get(key), label=key)
 
-                association_specs = [
+                lower = revised.casefold()
+                selected = None
+                specs = [
                     ("indirect", results.get("indirect"), [constructs.get("predictor"), constructs.get("mediator"), constructs.get("outcome")]),
                     ("direct", results.get("direct"), [constructs.get("predictor"), constructs.get("outcome")]),
                     ("a", results.get("a"), [constructs.get("predictor"), constructs.get("mediator")]),
                     ("b", results.get("b"), [constructs.get("mediator"), constructs.get("outcome")]),
                 ]
-                sentences = cls._sentence_spans(value)
-                metric_spans = []
-                for start, end, sentence in sentences:
-                    lower = sentence.casefold()
-                    selected = None
-                    if "indirect" in lower and all(str(x or "").casefold() in lower for x in association_specs[0][2] if x):
-                        selected = association_specs[0]
-                    elif (
-                        f"controlling {str(constructs.get('mediator') or '').casefold()}" in lower
-                        and str(constructs.get("predictor") or "").casefold() in lower
-                        and str(constructs.get("outcome") or "").casefold() in lower
-                    ):
-                        selected = association_specs[1]
-                    elif (
-                        f"controlling {str(constructs.get('predictor') or '').casefold()}" in lower
-                        and str(constructs.get("mediator") or "").casefold() in lower
-                        and str(constructs.get("outcome") or "").casefold() in lower
-                    ):
-                        selected = association_specs[3]
-                    elif str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() not in lower:
-                        selected = association_specs[2]
-                    elif str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower and str(constructs.get("predictor") or "").casefold() not in lower:
-                        selected = association_specs[3]
-                    if selected and selected[1] is not None:
-                        expected = float(selected[1])
-                        def beta_repl(match):
-                            observed = float(match.group(2))
-                            if abs(observed - expected) < 0.0006:
-                                return match.group(0)
-                            expected_text = f"{expected:.3f}".rstrip("0").rstrip(".")
-                            diagnostics.append({"type":"quantitative_metric_repaired","metric":selected[0],"observed":match.group(2),"expected":expected_text})
-                            return match.group(1) + expected_text
-                        revised = re.sub(r"(?i)((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", beta_repl, sentence)
-                        if revised != sentence:
-                            metric_spans.append((start, end, revised))
-                for start, end, replacement in sorted(metric_spans, reverse=True):
-                    value = value[:start] + replacement + value[end:]
+                if "indirect" in lower and all(str(x or "").casefold() in lower for x in specs[0][2] if x):
+                    selected = specs[0]
+                elif f"controlling {str(constructs.get('mediator') or '').casefold()}" in lower and str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower:
+                    selected = specs[1]
+                elif f"controlling {str(constructs.get('predictor') or '').casefold()}" in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower:
+                    selected = specs[3]
+                elif str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() not in lower:
+                    selected = specs[2]
+                elif str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower and str(constructs.get("predictor") or "").casefold() not in lower:
+                    selected = specs[3]
+                if selected and selected[1] is not None:
+                    replace_local_metric(r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", selected[1], label=selected[0])
+            elif not strict:
+                r2_map = dict(uploaded.get("r2_by_label") or {})
+                for label, expected in r2_map.items():
+                    if label and label in revised.casefold():
+                        replace_local_metric(rf"(R(?:²|\^2)\s*(?:for|of)\s*{re.escape(label)}\s*(?:=|:)\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_R2:{label}")
+                beta_map = dict(uploaded.get("beta_by_path") or {})
+                for path, expected in beta_map.items():
+                    left, _, right = path.partition(" -> ")
+                    if left and right and left in revised.casefold() and right in revised.casefold():
+                        replace_local_metric(r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_beta:{path}")
 
+            if revised != sentence:
+                replacements.append((start, end, revised))
+
+        for start, end, replacement in sorted(replacements, reverse=True):
+            value = value[:start] + replacement + value[end:]
         value = re.sub(r"[ \t]+\n", "\n", value)
         value = re.sub(r"\n{3,}", "\n\n", value)
         value = re.sub(r" {2,}", " ", value)