P0-R1.1-R1 Four-Language Dual-Engine Evidence Integrity

app.py

@@ -54,27 +54,23 @@ components.html(WECHAT_BLOCKER_JS, height=0, width=0)
 
 
 def inject_file_uploader_i18n(language: str) -> None:
-    """Localize Streamlit uploaders without modifying their React structure."""
+    """Localize uploaders by semantic label, never by visual/tab position."""
     payload = {
         "browse": tr(language, "browse_files"),
-        "uploaders": [
-            {
+        "uploaders": {
+            tr(language, "upload_reference"): {
                 "title": tr(language, "drop_docs"),
                 "limit": tr(language, "limit_docs"),
             },
-            {
-                "title": tr(language, "drop_data"),
-                "limit": tr(language, "limit_data"),
-            },
-            {
-                "title": tr(language, "drop_docs"),
-                "limit": tr(language, "limit_docs"),
-            },
-            {
+            tr(language, "upload_interviews"): {
                 "title": tr(language, "drop_interviews"),
                 "limit": tr(language, "limit_interviews"),
             },
-        ],
+            tr(language, "upload_stats"): {
+                "title": tr(language, "drop_data"),
+                "limit": tr(language, "limit_data"),
+            },
+        },
     }
 
     script = f"""
@@ -98,14 +94,22 @@ def inject_file_uploader_i18n(language: str) -> None:
         }}
       }}
 
+      function semanticUploaderConfig(uploader) {{
+        const label = uploader.querySelector('label');
+        const labelText = (label ? label.textContent : '').trim();
+        return payload.uploaders[labelText] || null;
+      }}
+
       function applyUploaderTranslations() {{
-        const zones = parentDocument.querySelectorAll(
-          '[data-testid="stFileUploaderDropzone"]'
+        const uploaders = parentDocument.querySelectorAll(
+          '[data-testid="stFileUploader"]'
         );
 
-        zones.forEach((zone, index) => {{
-          const config = payload.uploaders[index];
+        uploaders.forEach((uploader) => {{
+          const config = semanticUploaderConfig(uploader);
           if (!config) return;
+          const zone = uploader.querySelector('[data-testid="stFileUploaderDropzone"]');
+          if (!zone) return;
 
           const instructions = zone.querySelector(
             '[data-testid="stFileUploaderDropzoneInstructions"]'

core/generation_runner.py

@@ -901,6 +901,34 @@ class GenerationRunner:
         match = re.search(rf'(?m)^#\s+{re.escape(str(first_section_title))}\s*$', value)
         return value[:match.start()].strip() if match else ''
 
+    @staticmethod
+    def _extract_body_from_first_section(manuscript: str, first_section_title: str) -> str:
+        value = str(manuscript or '')
+        match = re.search(rf'(?m)^#\s+{re.escape(str(first_section_title))}\s*$', value)
+        return value[match.start():].strip() if match else value.strip()
+
+    @staticmethod
+    def _repair_validate_study_fact_contract(
+        mode: str, agent, quant_engine, text: str, data_metadata: dict, *, section_name: str
+    ):
+        value = str(text or '').strip()
+        diagnostics = []
+        if str(mode or '') == 'asq':
+            value, diagnostics = agent.repair_qualitative_study_fact_contract(
+                value, data_metadata, section_name=section_name
+            )
+            agent.validate_qualitative_study_fact_contract(
+                value, data_metadata, section_name=section_name
+            )
+        elif str(mode or '') == 'amj':
+            value, diagnostics = quant_engine.repair_quantitative_study_fact_contract(
+                value, data_metadata, section_name=section_name
+            )
+            quant_engine.validate_quantitative_study_fact_contract(
+                value, data_metadata, section_name=section_name
+            )
+        return value, list(diagnostics or [])
+
     @staticmethod
     def _replace_keywords(manuscript: str, keywords) -> str:
         value = str(manuscript or '')
@@ -1133,6 +1161,15 @@ class GenerationRunner:
         source_hash = identity['source_hash']
         compiled_source_hash = identity['compiled_source_hash']
         fact_ledger_hash = identity['fact_ledger_hash']
+        entity_ledger_method = getattr(agent, 'translation_protected_entity_ledger_hash', None)
+        protected_entity_ledger_hash = str(
+            entity_ledger_method(identity['compiled_source'], self._translation_worker_glossary)
+            if callable(entity_ledger_method) else ''
+        )
+        protected_entity_ledger_version = str(
+            getattr(agent, 'PROTECTED_ENTITY_LEDGER_VERSION', 'protected-entity-ledger-v1')
+            or 'protected-entity-ledger-v1'
+        )
         existing = dict((translation_sections or {}).get(unit_key) or {})
         if not isinstance(getattr(self, '_translation_progress_queued', None), set):
             self._translation_progress_queued = set()
@@ -1149,6 +1186,8 @@ class GenerationRunner:
             and existing.get('target_lang') == str(self._translation_worker_context.get('target_lang') or '')
             and existing.get('translation_contract_version') == TRANSLATION_CONTRACT_VERSION
             and existing.get('fact_ledger_hash') == fact_ledger_hash
+            and existing.get('protected_entity_ledger_version') == protected_entity_ledger_version
+            and existing.get('protected_entity_ledger_hash') == protected_entity_ledger_hash
             and existing.get('translated_text')
         ):
             return False
@@ -1157,11 +1196,13 @@ class GenerationRunner:
             if existing.get('source_hash') != source_hash: reasons.append('english_source_hash')
             if existing.get('compiled_source_hash') != compiled_source_hash: reasons.append('compiled_source_hash')
             if existing.get('fact_ledger_hash') != fact_ledger_hash: reasons.append('fact_ledger_hash')
+            if existing.get('protected_entity_ledger_version') != protected_entity_ledger_version: reasons.append('protected_entity_ledger_version')
+            if existing.get('protected_entity_ledger_hash') != protected_entity_ledger_hash: reasons.append('protected_entity_ledger_hash')
             if existing.get('translation_contract_version') != TRANSLATION_CONTRACT_VERSION: reasons.append('contract_version')
             if existing.get('target_lang') != str(self._translation_worker_context.get('target_lang') or ''): reasons.append('target_lang')
             if existing.get('qa_status') not in {'passed', 'passed_with_warning'}: reasons.append('qa_status')
             logging.info('Translation reuse miss | unit=%s phase=enqueue reasons=%s', unit_name, ','.join(reasons) or 'unknown')
-        pending_key = (str(unit_key), source_hash, compiled_source_hash, fact_ledger_hash)
+        pending_key = (str(unit_key), source_hash, compiled_source_hash, fact_ledger_hash, protected_entity_ledger_version, protected_entity_ledger_hash)
         if pending_key in self._translation_worker_pending:
             return False
         self._translation_progress_inflight.add(str(unit_key))
@@ -1241,6 +1282,8 @@ class GenerationRunner:
             and existing.get('target_lang') == target_lang
             and existing.get('translation_contract_version') == contract_version
             and existing.get('fact_ledger_hash') == fact_ledger_hash
+            and existing.get('protected_entity_ledger_version') == protected_entity_ledger_version
+            and existing.get('protected_entity_ledger_hash') == protected_entity_ledger_hash
             and existing.get('translated_text')
         ):
             self._translation_progress_inflight.discard(str(unit_key))
@@ -1250,6 +1293,8 @@ class GenerationRunner:
             if existing.get('source_hash') != source_hash: reasons.append('english_source_hash')
             if existing.get('compiled_source_hash') != compiled_source_hash: reasons.append('compiled_source_hash')
             if existing.get('fact_ledger_hash') != fact_ledger_hash: reasons.append('fact_ledger_hash')
+            if existing.get('protected_entity_ledger_version') != protected_entity_ledger_version: reasons.append('protected_entity_ledger_version')
+            if existing.get('protected_entity_ledger_hash') != protected_entity_ledger_hash: reasons.append('protected_entity_ledger_hash')
             if existing.get('translation_contract_version') != contract_version: reasons.append('contract_version')
             if existing.get('target_lang') != target_lang: reasons.append('target_lang')
             if existing.get('qa_status') not in {'passed', 'passed_with_warning'}: reasons.append('qa_status')
@@ -2009,6 +2054,23 @@ class GenerationRunner:
             working_keywords = list(working_front.get("keywords") or [])
             if not working_abstract or not working_keywords:
                 raise RuntimeError("working_front_matter_missing")
+            working_abstract, front_fact_rows = self._repair_validate_study_fact_contract(
+                mode, agent, quant_engine, working_abstract, data_metadata,
+                section_name="Working Abstract",
+            )
+            if not working_abstract:
+                raise RuntimeError("working_front_matter_empty_after_study_fact_contract")
+            if front_fact_rows:
+                logging.warning(
+                    "Working Abstract study-fact repair | mode=%s count=%s types=%s",
+                    mode, len(front_fact_rows),
+                    sorted({str(row.get("type") or "") for row in front_fact_rows}),
+                )
+                quality_diagnostics.append({
+                    "scope":"working_abstract_study_fact_integrity", "name":"Working Abstract",
+                    "status":"deterministic_repair", "count":len(front_fact_rows),
+                    "examples":front_fact_rows[:3],
+                })
             agent.global_memory["working_abstract"] = working_abstract
 
             self._progress(14, "outline_planning")
@@ -2230,6 +2292,26 @@ class GenerationRunner:
             state['canonical_concept_ledger'] = canonical_concept_ledger
         working_abstract = str(state.get("working_abstract") or "").strip()
         working_keywords = list(state.get("working_keywords") or [])
+        if working_abstract and not state.get("english_finalized"):
+            checked_working_abstract, resumed_front_rows = self._repair_validate_study_fact_contract(
+                mode, agent, quant_engine, working_abstract, data_metadata,
+                section_name="Working Abstract",
+            )
+            if not checked_working_abstract:
+                raise RuntimeError("working_front_matter_empty_after_study_fact_contract")
+            if checked_working_abstract != working_abstract or resumed_front_rows:
+                working_abstract = checked_working_abstract
+                state["working_abstract"] = working_abstract
+                if str(state.get("full_draft_en") or "").strip():
+                    state["full_draft_en"] = replace_abstract(str(state.get("full_draft_en") or ""), working_abstract)
+                    full_draft_en = str(state["full_draft_en"] or "")
+                quality_diagnostics.append({
+                    "scope":"working_abstract_study_fact_integrity", "name":"Working Abstract",
+                    "status":"resume_deterministic_repair", "count":len(resumed_front_rows),
+                    "examples":resumed_front_rows[:3],
+                })
+                state["quality_diagnostics"] = quality_diagnostics
+                self._checkpoint("working_front_matter_fact_reconciled", state, 15)
         if working_abstract:
             agent.global_memory["working_abstract"] = working_abstract
         if self._preview_target_text == '':
@@ -2249,6 +2331,29 @@ class GenerationRunner:
             state['translation_glossary'] = translation_glossary
             state['canonical_concept_ledger'] = canonical_concept_ledger
             self._checkpoint('canonical_concept_ledger_ready', state, 15)
+
+        current_entity_ledger_version = str(
+            getattr(translation_agent, 'PROTECTED_ENTITY_LEDGER_VERSION', 'protected-entity-ledger-v1')
+            or 'protected-entity-ledger-v1'
+        )
+        stale_entity_units = [
+            key for key, record in translation_sections.items()
+            if isinstance(record, dict)
+            and record.get('translated_text')
+            and record.get('qa_status') in {'passed', 'passed_with_warning'}
+            and record.get('protected_entity_ledger_version') != current_entity_ledger_version
+        ]
+        if stale_entity_units:
+            logging.warning(
+                'Protected Entity Ledger contract advanced; invalidating target-language cache only | units=%s version=%s',
+                stale_entity_units[:8], current_entity_ledger_version,
+            )
+            state['translation_complete'] = False
+            state['package_validated'] = False
+            state['full_draft_translated'] = ''
+            self._preview_target_text = ''
+            self._checkpoint('protected_entity_ledger_advanced', state, 15)
+
         self._start_translation_worker(
             translation_agent,
             translation_sections=translation_sections,
@@ -2277,6 +2382,23 @@ class GenerationRunner:
             working_keywords = list(working_front.get("keywords") or [])
             if not working_abstract or not working_keywords:
                 raise RuntimeError("working_front_matter_missing")
+            working_abstract, front_fact_rows = self._repair_validate_study_fact_contract(
+                mode, agent, quant_engine, working_abstract, data_metadata,
+                section_name="Working Abstract",
+            )
+            if not working_abstract:
+                raise RuntimeError("working_front_matter_empty_after_study_fact_contract")
+            if front_fact_rows:
+                logging.warning(
+                    "Working Abstract resume study-fact repair | mode=%s count=%s types=%s",
+                    mode, len(front_fact_rows),
+                    sorted({str(row.get("type") or "") for row in front_fact_rows}),
+                )
+                quality_diagnostics.append({
+                    "scope":"working_abstract_study_fact_integrity", "name":"Working Abstract",
+                    "status":"deterministic_repair", "count":len(front_fact_rows),
+                    "examples":front_fact_rows[:3],
+                })
             agent.global_memory["working_abstract"] = working_abstract
             full_draft_en = replace_abstract(full_draft_en, working_abstract)
             full_draft_en = self._replace_keywords(full_draft_en, working_keywords)
@@ -2685,6 +2807,24 @@ Output only final academic prose. Do not report the word count."""
                 agent.validate_qualitative_study_fact_contract(
                     en_resolved, data_metadata, section_name=display_name
                 )
+            elif mode == "amj":
+                en_resolved, study_fact_rows = quant_engine.repair_quantitative_study_fact_contract(
+                    en_resolved, data_metadata, section_name=display_name
+                )
+                if study_fact_rows:
+                    logging.warning(
+                        "Quantitative Evidence Ledger repair | unit=%s count=%s types=%s",
+                        display_name, len(study_fact_rows),
+                        sorted({str(row.get("type") or "") for row in study_fact_rows}),
+                    )
+                    quality_diagnostics.append({
+                        "scope":"quantitative_study_fact_integrity", "name":display_name,
+                        "status":"deterministic_repair", "count":len(study_fact_rows),
+                        "examples":study_fact_rows[:3],
+                    })
+                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
             )
@@ -2777,14 +2917,28 @@ Output only final academic prose. Do not report the word count."""
         if not state.get("english_finalized"):
             self._progress(91, "english_validation")
             full_draft_en = _clean_trailing_references(full_draft_en, ["Reference", "References", "Reference List"])
+            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(
-                    full_draft_en, data_metadata, section_name="Final English manuscript"
+                    final_body_for_fact, data_metadata, section_name="Final English body"
                 )
-                if fact_checked.strip() != full_draft_en.strip() or fact_rows:
+                if fact_checked.strip() != final_body_for_fact.strip() or fact_rows:
                     raise RuntimeError(
                         "final_qualitative_study_fact_contract_requires_section_regeneration"
                     )
+            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"
+                )
+                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"
             )
@@ -2897,6 +3051,23 @@ Output only final academic prose. Do not report the word count."""
             final_keywords = list(final_front.get("keywords") or [])
             if not final_abstract or not final_keywords:
                 raise RuntimeError("final_front_matter_missing")
+            final_abstract, final_front_fact_rows = self._repair_validate_study_fact_contract(
+                mode, agent, quant_engine, final_abstract, data_metadata,
+                section_name="Final Abstract",
+            )
+            if not final_abstract:
+                raise RuntimeError("final_abstract_empty_after_study_fact_contract")
+            if final_front_fact_rows:
+                logging.warning(
+                    "Final Abstract study-fact repair | mode=%s count=%s types=%s",
+                    mode, len(final_front_fact_rows),
+                    sorted({str(row.get("type") or "") for row in final_front_fact_rows}),
+                )
+                quality_diagnostics.append({
+                    "scope":"final_abstract_study_fact_integrity", "name":"Final Abstract",
+                    "status":"deterministic_repair", "count":len(final_front_fact_rows),
+                    "examples":final_front_fact_rows[:3],
+                })
             full_draft_en = replace_abstract(full_draft_en, final_abstract)
             full_draft_en = self._replace_keywords(full_draft_en, final_keywords)
             body_only, _reference_tail = agent.validate_reference_tail(

core/pipeline_agent.py

@@ -72,7 +72,7 @@ class CitationAuditError(DeepSeekRequestError):
 class SSciPipelineAgent:
     TRANSLATION_CONTRACT_VERSION = 'hierarchical-multilingual-v8'
     MAX_TRANSLATION_CLAIM_REPAIR_ATTEMPTS = 2
-    PROTECTED_ENTITY_LEDGER_VERSION = 'protected-entity-ledger-v1'
+    PROTECTED_ENTITY_LEDGER_VERSION = 'protected-entity-ledger-v2'
 
     def __init__(self, ds_key, s2_key=""):
         self.ds_key = str(ds_key or "").strip()
@@ -4676,6 +4676,32 @@ Output only the repaired section."""
                 if cls._translation_technical_entity_shape(token):
                     add(token, 'MODEL_TOOL', claim_id, token)
 
+            # Source-owned multi-word proper names remain legal even at the
+            # beginning of a claim (for example ``Krishi Saathi`` or
+            # ``Stack Overflow``).  This closes the v1 sentence-start blind spot
+            # without authorizing ordinary sentence-initial words.
+            multiword_pattern = re.compile(
+                r'(?<![A-Za-z0-9_-])('                 r'[A-Z][A-Za-z0-9+.#-]{2,}'                 r'(?:\s+(?:(?:of|the|and|for|in|at|on|by|to|with)\s+)?[A-Z][A-Za-z0-9+.#-]{2,}){1,5}'                 r')(?![A-Za-z0-9_-])'
+            )
+            for match in multiword_pattern.finditer(claim_source):
+                phrase = re.sub(r'\s+', ' ', match.group(1)).strip()
+                first = phrase.split()[0].casefold() if phrase else ''
+                if first in ordinary_sentence_starters and len(phrase.split()) < 2:
+                    continue
+                add(phrase, 'PROPER_NAME', claim_id, phrase)
+
+            # A sentence-initial Title-Case technical term is protected only
+            # when its immediate source context names a bounded technical role.
+            # Ordinary words such as ``Sensitizing`` do not qualify.
+            technical_context = re.compile(
+                r'(?<![A-Za-z0-9_-])([A-Z][A-Za-z0-9+.#-]{3,})\s+'
+                r'(model|architecture|framework|network|encoder|decoder|algorithm|platform|system)\b'
+            )
+            for match in technical_context.finditer(claim_source):
+                token = match.group(1)
+                if token.casefold() not in ordinary_sentence_starters:
+                    add(token, 'TECHNICAL_TERM', claim_id, token)
+
             # Conservative standalone proper-name support.  Hyphen-adjacent words
             # are excluded here and handled only by the bounded compound rule above,
             # preventing ``Open-ended`` from authorizing a bare ``Open``.

core/quantitative_engine.py

@@ -17,6 +17,8 @@ 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"
+
     @staticmethod
     def generate_psd_corr_matrix(n_constructs, max_corr=0.5):
         if n_constructs == 1:
@@ -302,6 +304,306 @@ Keywords: {keywords}"""
         }
         return summary, path_table, spearman
 
+
+    @staticmethod
+    def _sentence_spans(text):
+        """Return sentence spans without splitting decimal statistics or citations."""
+        value = str(text or "")
+        spans = []
+        start = 0
+        round_depth = 0
+        square_depth = 0
+        index = 0
+        abbreviations = ("e.g.", "i.e.", "et al.", "fig.", "dr.", "mr.", "ms.", "prof.", "vs.", "no.", "vol.")
+        while index < len(value):
+            char = value[index]
+            if char == "(": round_depth += 1
+            elif char == ")" and round_depth: round_depth -= 1
+            elif char == "[": square_depth += 1
+            elif char == "]" and square_depth: square_depth -= 1
+            boundary = False
+            if char == "\n":
+                boundary = True
+            elif char in ".!?。！？" and round_depth == 0 and square_depth == 0:
+                if char == ".":
+                    prev = value[index - 1] if index > 0 else ""
+                    nxt = value[index + 1] if index + 1 < len(value) else ""
+                    decimal = prev.isdigit() and nxt.isdigit()
+                    prefix = value[max(0, index - 12): index + 1].lower()
+                    abbreviation = any(prefix.endswith(item) for item in abbreviations)
+                    boundary = not decimal and not abbreviation
+                else:
+                    boundary = True
+            if boundary:
+                end = index + (0 if char == "\n" else 1)
+                if end > start and value[start:end].strip():
+                    spans.append((start, end, value[start:end]))
+                start = index + 1
+            index += 1
+        if start < len(value) and value[start:].strip():
+            spans.append((start, len(value), value[start:]))
+        return spans
+
+    @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.
+        """
+        meta = dict(metadata or {})
+        source_type = str(meta.get("source_type") or "")
+        sample_size = meta.get("sample_size")
+        scale_points = meta.get("scale_points")
+        strict = source_type == "system_generated"
+        evidence = str(evidence_text or "")
+        if not strict and evidence:
+            sample_candidates = set()
+            for pattern in (
+                r"(?i)\b(?:sample\s+size|sample|N)\s*(?:=|:|of)?\s*(\d{2,6})\b",
+                r"(?i)\b(\d{2,6})\s+(?:respondents|participants|observations|cases)\b",
+            ):
+                sample_candidates.update(int(x) for x in re.findall(pattern, evidence))
+            if len(sample_candidates) == 1:
+                sample_size = next(iter(sample_candidates))
+            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))
+        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 {
+            "version": cls.QUANTITATIVE_EVIDENCE_LEDGER_VERSION,
+            "source_type": source_type,
+            "strict": bool(strict),
+            "sample_size": int(sample_size) if isinstance(sample_size, (int, np.integer)) or str(sample_size or "").isdigit() else None,
+            "scale_points": int(scale_points) if isinstance(scale_points, (int, np.integer)) or str(scale_points or "").isdigit() else None,
+            "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,
+            "causal_identification": "none" if strict else "evidence_owned",
+            "allowed_methods": [
+                "reliability", "composite reliability", "average variance extracted",
+                "Fornell-Larcker", "standardized association", "bootstrap",
+                "Spearman", "VIF", "descriptive statistics",
+            ] if strict else [],
+            "forbidden_unowned_methods": [
+                "confirmatory factor analysis", "CFA", "structural equation modeling", "SEM",
+                "PLS-SEM", "RMSEA", "CFI", "TLI", "SRMR", "Harman",
+                "common method bias", "endogeneity", "instrumental variable",
+                "fixed effects", "random effects", "difference-in-differences",
+            ] if strict else [],
+        }
+
+    @classmethod
+    def repair_quantitative_study_fact_contract(cls, text, metadata, section_name=""):
+        """Repair only deterministic AMJ study-fact drift owned by frozen evidence."""
+        value = str(text or "")
+        meta = dict(metadata or {})
+        ledger = dict(meta.get("quantitative_evidence_ledger") or {})
+        if not ledger:
+            ledger = cls.build_quantitative_evidence_ledger(meta)
+        diagnostics = []
+        sample_size = ledger.get("sample_size")
+        scale_points = ledger.get("scale_points")
+        strict = bool(ledger.get("strict"))
+        constructs = dict(ledger.get("constructs") or {})
+        results = dict(ledger.get("association_results") or {})
+
+        # 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:
+                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):
+                    observed = int(match.group(2))
+                    if observed == int(sample_size):
+                        return match.group(0)
+                    diagnostics.append({"type":"sample_size_repaired","observed":observed,"expected":int(sample_size)})
+                    return match.group(1) + str(int(sample_size))
+                revised = sample_pattern.sub(sample_repl, revised)
+                count_pattern = re.compile(r"(?i)\b(\d{2,6})\s+(respondents|participants|observations|cases)\b")
+                def count_repl(match):
+                    observed = int(match.group(1))
+                    if observed == int(sample_size):
+                        return match.group(0)
+                    diagnostics.append({"type":"sample_count_repaired","observed":observed,"expected":int(sample_size)})
+                    return f"{int(sample_size)} {match.group(2)}"
+                revised = count_pattern.sub(count_repl, revised)
+            if scale_points:
+                scale_pattern = re.compile(r"(?i)\b([2-9]|1[0-1])[- ]point\s+(Likert(?:-type)?\s+scale)")
+                def scale_repl(match):
+                    observed = int(match.group(1))
+                    if observed == int(scale_points):
+                        return match.group(0)
+                    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():
+                    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)
+                    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:
+                    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 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
+                if expected is None:
+                    return
+                expected_text = str(int(expected)) if dtype == "int" else f"{float(expected):.3f}".rstrip("0").rstrip(".")
+                regex = re.compile(pattern, re.I)
+                def repl(match):
+                    observed = match.group(2)
+                    try:
+                        same = int(float(observed)) == int(expected) if dtype == "int" else abs(float(observed) - float(expected)) < 0.0006
+                    except Exception:
+                        same = False
+                    if same:
+                        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)
+
+            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")
+                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)
+
+                association_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:]
+
+        value = re.sub(r"[ \t]+\n", "\n", value)
+        value = re.sub(r"\n{3,}", "\n\n", value)
+        value = re.sub(r" {2,}", " ", value)
+        return value.strip(), diagnostics
+
+    @classmethod
+    def validate_quantitative_study_fact_contract(cls, text, metadata, section_name=""):
+        repaired, diagnostics = cls.repair_quantitative_study_fact_contract(text, metadata, section_name=section_name)
+        if repaired.strip() != str(text or "").strip():
+            raise RuntimeError(f"quantitative_study_fact_contract_unrepaired:{section_name}:{diagnostics[:3]}")
+        return True
+
     def resolve_empirical_data(
         self,
         uploaded_files=None,
@@ -333,6 +635,7 @@ Keywords: {keywords}"""
                         "Do not infer absent setting, sample, measures, software, or results."
                     ),
                 }
+                metadata["quantitative_evidence_ledger"] = self.build_quantitative_evidence_ledger(metadata, user_md)
                 return None, user_md, tr(language, "actual_stats_parsed"), metadata
 
         design = self._derive_system_design(agent, topic, keywords)
@@ -433,6 +736,7 @@ Keywords: {keywords}"""
                 "Frozen bootstrap and collinearity checks",
             ],
         }
+        metadata["quantitative_evidence_ledger"] = self.build_quantitative_evidence_ledger(metadata, generated_prompt)
         return (
             df,
             generated_prompt,