Blame for core/quantitative_engine.py

# -*- coding: utf-8 -*-
"""Quantitative evidence preparation for SmartPaper's AMJ-oriented track."""

from __future__ import annotations

import hashlib
import json
import logging
import re

import numpy as np
import pandas as pd

from i18n import tr
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-v2"
    QUANTITATIVE_TABLE_MANIFEST_VERSION = "quantitative-table-manifest-v1"
    HYPOTHESIS_LEDGER_VERSION = "quantitative-hypothesis-ledger-v1"
    TARGET_TABLE_BINDING_VERSION = "target-quantitative-table-binding-v1"
    SEMANTIC_CLAIM_FRAME_VERSION = "smartpaper-core-semantic-claim-frame-v1"

    _TARGET_TABLE_LOCALIZATION = {
        "zh": {
            "caption": "表 {number}：{title}",
            "titles": {
                1: "信度与收敛效度",
                2: "描述性统计与判别效度（Fornell-Larcker 准则）",
                3: "关联模型结果",
                4: "秩次稳健性相关（Spearman）",
            },
            "cells": {
                "Construct": "构念", "Items": "题项数", "Cronbach's α": "Cronbach α",
                "Mean": "均值", "SD": "SD", "Association": "关联关系",
                "Standardized estimate": "标准化估计", "R² of outcome": "结果变量的 R²",
                "Path": "路径", "Beta": "Beta", "Estimate": "估计量",
                "Point estimate": "点估计", "95% bootstrap CI lower": "95% bootstrap CI 下限",
                "95% bootstrap CI upper": "95% bootstrap CI 上限",
            },
            "phrases": {
                "Indirect association:": "间接关联：",
                "Indirect association": "间接关联",
                "Residual direct association": "剩余直接关联",
                "controlling": "控制",
            },
            "notes": {
                "Diagonal elements in parentheses are the square roots of AVE.": "括号中的对角线元素为 AVE 的平方根。",
            },
            "note_prefix": "注：",
        },
        "ru": {
            "caption": "Таблица {number}: {title}",
            "titles": {
                1: "Надёжность и конвергентная валидность",
                2: "Описательная статистика и дискриминантная валидность (критерий Форнелла–Ларкера)",
                3: "Результаты модели ассоциаций",
                4: "Ранговые корреляции для проверки устойчивости (Спирмен)",
            },
            "cells": {
                "Construct": "Конструкт", "Items": "Пункты", "Cronbach's α": "α Кронбаха",
                "Mean": "Среднее", "SD": "SD", "Association": "Ассоциация",
                "Standardized estimate": "Стандартизированная оценка", "R² of outcome": "R² результата",
                "Path": "Связь", "Beta": "Beta", "Estimate": "Оценка",
                "Point estimate": "Точечная оценка", "95% bootstrap CI lower": "Нижняя граница 95% bootstrap CI",
                "95% bootstrap CI upper": "Верхняя граница 95% bootstrap CI",
            },
            "phrases": {
                "Indirect association:": "Косвенная ассоциация:",
                "Indirect association": "Косвенная ассоциация",
                "Residual direct association": "Остаточная прямая ассоциация",
                "controlling": "при контроле",
            },
            "notes": {
                "Diagonal elements in parentheses are the square roots of AVE.": "Диагональные элементы в скобках представляют квадратные корни из AVE.",
            },
            "note_prefix": "Примечание:",
        },
        "kk": {
            "caption": "{number}-кесте: {title}",
            "titles": {
                1: "Сенімділік және конвергенттік валидтілік",
                2: "Сипаттамалық статистика және дискриминанттық валидтілік (Форнелл–Ларкер критерийі)",
                3: "Байланыстар моделінің нәтижелері",
                4: "Орнықтылықты тексеруге арналған рангтік корреляциялар (Спирмен)",
            },
            "cells": {
                "Construct": "Конструкт", "Items": "Тармақтар", "Cronbach's α": "Кронбах α",
                "Mean": "Орташа", "SD": "SD", "Association": "Байланыс",
                "Standardized estimate": "Стандартталған бағалау", "R² of outcome": "Нәтиженің R²",
                "Path": "Байланыс", "Beta": "Beta", "Estimate": "Бағалау",
                "Point estimate": "Нүктелік бағалау", "95% bootstrap CI lower": "95% bootstrap CI төменгі шегі",
                "95% bootstrap CI upper": "95% bootstrap CI жоғарғы шегі",
            },
            "phrases": {
                "Indirect association:": "Жанама байланыс:",
                "Indirect association": "Жанама байланыс",
                "Residual direct association": "Қалдық тікелей байланыс",
                "controlling": "бақылау кезінде",
            },
            "notes": {
                "Diagonal elements in parentheses are the square roots of AVE.": "Жақшадағы диагональ элементтер AVE квадрат түбірлерін көрсетеді.",
            },
            "note_prefix": "Ескертпе:",
        },
    }

    @classmethod
    def build_hypothesis_ledger(cls, constructs):
        """Freeze one result-independent AMJ hypothesis identity contract.

        The system-generated quantitative design is observational and does not
        justify result-driven directional hypotheses.  The three numbered
        hypotheses therefore freeze only the construct pairing/association
        identity.  The indirect association is an analysis target, not H4.
        """
        values = [re.sub(r"\s+", " ", str(item or "")).strip() for item in (constructs or [])]
        if len(values) != 3 or any(not item for item in values):
            return {}
        predictor, mediator, outcome = values
        items = [
            {"hypothesis_id": "H1", "predictor": predictor, "outcome": outcome, "relation": "association"},
            {"hypothesis_id": "H2", "predictor": predictor, "outcome": mediator, "relation": "association"},
            {"hypothesis_id": "H3", "predictor": mediator, "outcome": outcome, "relation": "association"},
        ]
        return {
            "version": cls.HYPOTHESIS_LEDGER_VERSION,
            "items": items,
            "indirect_analysis": {
                "predictor": predictor,
                "mediator": mediator,
                "outcome": outcome,
                "numbered_hypothesis": False,
                "relation": "estimated_indirect_association",
            },
        }

    @classmethod
    def build_table_manifest_from_evidence_markdown(cls, evidence_markdown):
        """Rebuild the synthetic table manifest from a durable pre-C5 evidence packet."""
        value = str(evidence_markdown or "")
        targets = {
            1: "Descriptive Statistics and Measurement Assessment",
            2: "Descriptive Statistics and Measurement Assessment",
            3: "Hypothesis and Association Tests",
            4: "Additional Analyses and Robustness Checks",
        }
        rows = []
        pattern = re.compile(
            r"(?ms)^###\s+Table\s+(?P<number>[1-4])\s*:\s*(?P<title>[^\n]+)\n(?P<body>.*?)(?=^###\s+|\Z)"
        )
        for match in pattern.finditer(value):
            number = int(match.group("number"))
            body = match.group("body").strip()
            table_lines = []
            note = ""
            for line in body.splitlines():
                stripped = line.strip()
                if stripped.startswith("|"):
                    table_lines.append(line.rstrip())
                elif re.match(r"^\*?Note\s*:", stripped, flags=re.I):
                    note = re.sub(r"^\*?Note\s*:\s*", "", stripped, flags=re.I).strip("* ")
                elif table_lines:
                    break
            if len(table_lines) < 2:
                continue
            title = re.sub(r"^Frozen\s+", "", match.group("title").strip(), flags=re.I)
            row = {
                "number": number,
                "title": title,
                "target_subsection": targets[number],
                "markdown": "\n".join(table_lines),
            }
            if note:
                row["note"] = note
            rows.append(row)
        if {row["number"] for row in rows} != {1, 2, 3, 4}:
            return {}
        return {"version": cls.QUANTITATIVE_TABLE_MANIFEST_VERSION, "tables": sorted(rows, key=lambda row: row["number"])}

    @classmethod
    def build_table_manifest(cls, validity_df, fl_matrix_df, path_table, spearman):
        """Return the exact reader-facing tables required by the synthetic AMJ path."""
        rows = [
            {
                "number": 1,
                "title": "Reliability and Convergent Validity",
                "target_subsection": "Descriptive Statistics and Measurement Assessment",
                "markdown": validity_df.to_markdown(index=False),
            },
            {
                "number": 2,
                "title": "Descriptive Statistics and Discriminant Validity (Fornell-Larcker Criterion)",
                "target_subsection": "Descriptive Statistics and Measurement Assessment",
                "note": "Diagonal elements in parentheses are the square roots of AVE.",
                "markdown": fl_matrix_df.to_markdown(),
            },
            {
                "number": 3,
                "title": "Association Model Results",
                "target_subsection": "Hypothesis and Association Tests",
                "markdown": path_table.to_markdown(index=False),
            },
            {
                "number": 4,
                "title": "Rank-Order Robustness Correlations (Spearman)",
                "target_subsection": "Additional Analyses and Robustness Checks",
                "markdown": spearman.round(3).to_markdown(),
            },
        ]
        return {"version": cls.QUANTITATIVE_TABLE_MANIFEST_VERSION, "tables": rows}

    @staticmethod
    def _target_table_language_code(target_lang):
        value = str(target_lang or "").strip().casefold().replace("_", "-")
        if value.startswith("zh") or "chinese" in value or "中文" in value or "简体" in value:
            return "zh"
        if value.startswith("ru") or "russian" in value or "рус" in value:
            return "ru"
        if value.startswith("kk") or value.startswith("kz") or "kazakh" in value or "қазақ" in value:
            return "kk"
        return "en"

    @staticmethod
    def _replace_glossary_terms(value, glossary):
        text = str(value or "")
        pairs = []
        for source, target in dict(glossary or {}).items():
            source = re.sub(r"\s+", " ", str(source or "")).strip()
            target = re.sub(r"\s+", " ", str(target or "")).strip()
            if not source or not target or source == target:
                continue
            if not re.search(r"[A-Za-z]", source):
                continue
            pairs.append((source, target))
        for source, target in sorted(pairs, key=lambda row: len(row[0]), reverse=True):
            pattern = re.compile(
                rf"(?<![A-Za-z0-9_]){re.escape(source)}(?![A-Za-z0-9_])",
                flags=re.I,
            )
            text = pattern.sub(lambda _match, replacement=target: replacement, text)
        return text

    @classmethod
    def _localize_quantitative_table_cell(cls, value, target_lang, glossary=None):
        code = cls._target_table_language_code(target_lang)
        if code == "en":
            return cls._replace_glossary_terms(value, glossary)
        profile = cls._TARGET_TABLE_LOCALIZATION[code]
        text = str(value or "")
        stripped = text.strip()
        exact = dict(profile.get("cells") or {})
        if stripped in exact:
            return text.replace(stripped, str(exact[stripped]), 1)
        text = cls._replace_glossary_terms(text, glossary)
        for source, target in sorted(
            dict(profile.get("phrases") or {}).items(), key=lambda row: len(row[0]), reverse=True
        ):
            text = re.sub(re.escape(source), str(target), text, flags=re.I)
        return text

    @classmethod
    def _localize_quantitative_markdown_table(cls, markdown, target_lang, glossary=None):
        lines = []
        for line in str(markdown or "").splitlines():
            if not line.lstrip().startswith("|"):
                lines.append(line)
                continue
            raw = line.rstrip()
            cells = raw.split("|")
            # Alignment rows contain only punctuation/colons and must remain byte-stable.
            if all(not re.search(r"[A-Za-z\u0400-\u04FF\u4e00-\u9fff]", cell) for cell in cells):
                lines.append(raw)
                continue
            localized = [cls._localize_quantitative_table_cell(cell, target_lang, glossary) for cell in cells]
            lines.append("|".join(localized))
        return "\n".join(lines).strip()

    @classmethod
    def localize_quantitative_table_manifest(cls, manifest, target_lang, glossary=None):
        record = dict(manifest or {})
        rows = [dict(item) for item in (record.get("tables") or []) if isinstance(item, dict)]
        if {int(row.get("number") or 0) for row in rows} != {1, 2, 3, 4}:
            return {}
        code = cls._target_table_language_code(target_lang)
        profile = cls._TARGET_TABLE_LOCALIZATION.get(code, {})
        localized_rows = []
        for row in sorted(rows, key=lambda item: int(item.get("number") or 0)):
            number = int(row.get("number") or 0)
            title = str(row.get("title") or "").strip()
            note = str(row.get("note") or "").strip()
            if code != "en":
                title = str(dict(profile.get("titles") or {}).get(number) or title)
                note = str(dict(profile.get("notes") or {}).get(note) or note)
            localized_rows.append({
                **row,
                "number": number,
                "title": title,
                "markdown": cls._localize_quantitative_markdown_table(
                    row.get("markdown"), target_lang, glossary
                ),
                **({"note": note} if note else {}),
            })
        return {
            "version": cls.TARGET_TABLE_BINDING_VERSION,
            "source_manifest_version": str(record.get("version") or ""),
            "language": code,
            "tables": localized_rows,
        }

    @staticmethod
    def _strip_target_quantitative_markdown_tables(text):
        """Remove any model-rendered table blocks before manifest-owned rebinding."""
        lines = str(text or "").splitlines()
        kept = []
        index = 0
        caption_pattern = re.compile(
            r"^\s*(?:Table\s+\d+|表\s*\d+|Таблиц(?:а|ы|е|у|ей|ами|ах)?\s*\d+|Кесте\s*\d+|\d+\s*[-–—]?\s*кесте)\s*[:：.\-–—]",
            flags=re.I,
        )
        note_pattern = re.compile(r"^\s*\*?(?:Note|注|Примечание|Ескертпе)\s*[:：]", flags=re.I)
        while index < len(lines):
            line = lines[index]
            if caption_pattern.match(line):
                index += 1
                while index < len(lines) and not lines[index].strip():
                    index += 1
                if index < len(lines) and lines[index].lstrip().startswith("|"):
                    while index < len(lines) and lines[index].lstrip().startswith("|"):
                        index += 1
                while index < len(lines) and not lines[index].strip():
                    index += 1
                if index < len(lines) and note_pattern.match(lines[index]):
                    index += 1
                continue
            if line.lstrip().startswith("|"):
                start = index
                while index < len(lines) and lines[index].lstrip().startswith("|"):
                    index += 1
                block = lines[start:index]
                if len(block) >= 2 and re.search(r"-{3,}", block[1]):
                    continue
                kept.extend(block)
                continue
            kept.append(line)
            index += 1
        return re.sub(r"\n{3,}", "\n\n", "\n".join(kept)).strip()

    @classmethod
    def bind_target_language_quantitative_tables(cls, translated_results, manifest, target_lang, glossary=None):
        """Bind exactly Table 1-4 to one translated Results unit without model ownership.

        This operation is deterministic and idempotent. Statistical values remain
        source-manifest owned; only captions, generic statistical labels, notes,
        and canonical construct surfaces are localized.
        """
        localized = cls.localize_quantitative_table_manifest(manifest, target_lang, glossary)
        rows = [item for item in (localized.get("tables") or []) if isinstance(item, dict)]
        if {int(row.get("number") or 0) for row in rows} != {1, 2, 3, 4}:
            raise RuntimeError("target_quantitative_table_manifest_invalid")
        code = cls._target_table_language_code(target_lang)
        profile = cls._TARGET_TABLE_LOCALIZATION.get(code, {})
        base = cls._strip_target_quantitative_markdown_tables(translated_results)
        blocks = []
        for row in rows:
            number = int(row.get("number") or 0)
            title = str(row.get("title") or "").strip()
            markdown = str(row.get("markdown") or "").strip()
            if not number or not title or not markdown:
                raise RuntimeError(f"target_quantitative_table_manifest_invalid:Table {number}")
            if code == "zh":
                caption = f"表 {number}：{title}"
            elif code == "ru":
                caption = f"Таблица {number}: {title}"
            elif code == "kk":
                caption = f"{number}-кесте: {title}"
            else:
                caption = f"Table {number}: {title}"
            block = caption + "\n" + markdown
            note = str(row.get("note") or "").strip()
            if note:
                prefix = str(profile.get("note_prefix") or "Note:")
                block += f"\n\n*{prefix} {note}*"
            blocks.append(block)
        return (base.rstrip() + "\n\n" + "\n\n".join(blocks)).strip()

    @staticmethod
    def generate_psd_corr_matrix(n_constructs, max_corr=0.5):
        if n_constructs == 1:
            return np.array([[1.0]])
        base_corr = np.random.uniform(0.2, max_corr - 0.1)
        matrix = np.random.uniform(
            base_corr - 0.1,
            base_corr + 0.1,
            (n_constructs, n_constructs),
        )
        matrix = (matrix + matrix.T) / 2
        np.fill_diagonal(matrix, 1.0)
        min_eig = np.min(np.linalg.eigvals(matrix))
        if min_eig < 0:
            matrix -= (min_eig - 0.01) * np.eye(n_constructs)
        diagonal = np.sqrt(np.diag(matrix))
        return matrix / np.outer(diagonal, diagonal)

    @staticmethod
    def calculate_validity_metrics(df, constructs_info, true_loadings):
        metrics = []
        for construct in constructs_info:
            construct_name = construct["name"]
            items = construct["items"]
            sub_df = df[items]
            var_total = sub_df.sum(axis=1).var(ddof=1)
            sum_var = sub_df.var(ddof=1).sum()
            n_items = len(items)
            alpha = (
                0.0
                if var_total <= 0
                else (n_items / (n_items - 1))
                * (1 - sum_var / var_total)
            )
            loadings = np.array(true_loadings[construct_name])
            sum_loadings_sq = np.sum(loadings) ** 2
            sum_sq_loadings = np.sum(loadings**2)
            sum_error = np.sum(1 - loadings**2)
            cr = sum_loadings_sq / (sum_loadings_sq + sum_error)
            ave = sum_sq_loadings / len(loadings)
            metrics.append(
                {
                    "Construct": construct_name,
                    "Items": n_items,
                    "Cronbach's α": round(alpha, 3),
                    "CR": round(cr, 3),
                    "AVE": round(ave, 3),
                }
            )
        return pd.DataFrame(metrics)

    @staticmethod
    def _safe_constructs(values):
        cleaned = []
        for value in values or []:
            name = re.sub(r"\s+", " ", str(value or "")).strip(" .,:;-_\n\t")
            if not name or len(name) > 80 or name.lower() in {item.lower() for item in cleaned}:
                continue
            cleaned.append(name)
        return cleaned[:3]

    def _derive_system_design(self, agent, topic="", keywords=""):
        """Freeze a parsimonious topic-aligned design without inventing named sites."""
        fallback = {
            "constructs": [
                "Focal Predictor",
                "Focal Mediator",
                "Focal Outcome",
            ],
            "research_context": "respondents with experience relevant to the focal management phenomenon",
        }
        if agent is None:
            return fallback
        prompt = f"""Design the minimum coherent quantitative management-study skeleton for the topic below.
Return ONLY valid JSON:
{{"constructs":["predictor construct","mediating construct","outcome construct"],"research_context":"one generic respondent context"}}
Rules:
1. Exactly three distinct constructs in predictor -> mediator -> outcome order.
2. Use concise construct labels that can appear throughout a journal article.
3. The context must be generic and directly relevant to the topic. Do NOT name or invent a country, university, firm, industry sample frame, software package, time period, or institution.
4. This is a cross-sectional single-wave survey design, so do not imply experimental or causal identification.
5. Do not create any number; sample size and scale are frozen elsewhere.
Topic: {topic}
Keywords: {keywords}"""
        try:
            raw = agent.call_deepseek(
                "You are a rigorous management research-design editor.",
                prompt,
                max_tokens=500,
                temp=0.0,
                model_role="fast",
                thinking=False,
                allow_fallback=True,
            )
            parsed = agent._extract_json_from_text(raw)
            if not isinstance(parsed, dict):
                return fallback
            constructs = self._safe_constructs(parsed.get("constructs"))
            context = re.sub(r"\s+", " ", str(parsed.get("research_context") or "")).strip()
            if len(constructs) != 3:
                return fallback
            if not context or len(context) > 240:
                context = fallback["research_context"]
            # Named-location/site details are deliberately not accepted from this
            # design step. The manuscript may remain generic when no user evidence
            # supplies a real setting.
            if re.search(r"\b(university|company|corporation|inc\.?|ltd\.?|country|china|usa|united states|europe|asia)\b", context, re.I):
                context = fallback["research_context"]
            return {"constructs": constructs, "research_context": context}
        except Exception as exc:
            logging.warning("Quantitative study-design freeze fell back safely: %s", exc)
            return fallback

    def generate_empirical_data_robust(
        self,
        construct_names=None,
        n=300,
        scale_points=7,
    ):
        if not construct_names or len(construct_names) < 2:
            construct_names = [
                "Focal Predictor",
                "Focal Mediator",
                "Focal Outcome",
            ]

        n_constructs = len(construct_names)
        corr = self.generate_psd_corr_matrix(n_constructs, max_corr=0.55)
        latent_scores = np.random.multivariate_normal(
            mean=[0] * n_constructs,
            cov=corr,
            size=n,
        )
        data = {}
        true_loadings = {}
        constructs_info = []
        target_mean = scale_points / 2.0 + 0.5
        target_std = scale_points / 5.0

        for index, construct_name in enumerate(construct_names):
            true_score = latent_scores[:, index]
            n_items = np.random.randint(3, 6)
            loadings = np.random.uniform(0.72, 0.88, n_items)
            true_loadings[construct_name] = loadings
            item_names = []
            safe_prefix = re.sub(r"[^A-Za-z0-9]+", "_", construct_name).strip("_") or f"Construct_{index + 1}"
            for item_index, loading in enumerate(loadings):
                error = np.random.normal(
                    0,
                    np.sqrt(1 - loading**2),
                    n,
                )
                item_score = loading * true_score + error
                raw_scaled = item_score * target_std + target_mean
                item_int = np.clip(
                    np.round(raw_scaled),
                    1,
                    scale_points,
                ).astype(int)
                column_name = f"{safe_prefix}_{item_index + 1}"
                data[column_name] = item_int
                item_names.append(column_name)
            constructs_info.append(
                {"name": construct_name, "items": item_names}
            )

        df = pd.DataFrame(data)
        validity_df = self.calculate_validity_metrics(
            df,
            constructs_info,
            true_loadings,
        )
        mean_scores = pd.DataFrame(
            {
                construct["name"]: df[construct["items"]].mean(axis=1)
                for construct in constructs_info
            }
        )
        corr_matrix = mean_scores.corr()
        fl_matrix = corr_matrix.astype(object)

        for _, row in validity_df.iterrows():
            construct_name = row["Construct"]
            sqrt_ave = np.sqrt(row["AVE"])
            fl_matrix.loc[construct_name, construct_name] = (
                f"({sqrt_ave:.3f})"
            )

        for row_index in range(len(fl_matrix.columns)):
            for column_index in range(row_index + 1, len(fl_matrix.columns)):
                fl_matrix.iloc[row_index, column_index] = ""
            for column_index in range(row_index):
                value = fl_matrix.iloc[row_index, column_index]
                if isinstance(value, (int, float, np.integer, np.floating)):
                    fl_matrix.iloc[row_index, column_index] = f"{float(value):.3f}"

        desc_stats = mean_scores.mean().to_frame("Mean").join(
            mean_scores.std().to_frame("SD")
        )
        return df, validity_df, desc_stats.join(fl_matrix), mean_scores, corr_matrix

    @staticmethod
    def _standardized_ols(y, predictors):
        y = np.asarray(y, dtype=float)
        x = np.column_stack([np.asarray(value, dtype=float) for value in predictors])
        y_sd = y.std(ddof=1)
        x_sd = x.std(axis=0, ddof=1)
        if y_sd <= 0 or np.any(x_sd <= 0):
            raise RuntimeError("quantitative_zero_variance")
        yz = (y - y.mean()) / y_sd
        xz = (x - x.mean(axis=0)) / x_sd
        design = np.column_stack([np.ones(len(yz)), xz])
        coef = np.linalg.lstsq(design, yz, rcond=None)[0]
        fitted = design @ coef
        ss_res = float(np.sum((yz - fitted) ** 2))
        ss_tot = float(np.sum((yz - yz.mean()) ** 2))
        r2 = 0.0 if ss_tot <= 0 else 1.0 - ss_res / ss_tot
        return coef[1:], max(0.0, min(1.0, r2))

    def _path_evidence(self, mean_scores, construct_names, bootstrap_iterations=1500):
        if len(construct_names) < 3:
            return {}, pd.DataFrame(), pd.DataFrame()
        x_name, m_name, y_name = construct_names[0], construct_names[1], construct_names[2]
        x = mean_scores[x_name].to_numpy(dtype=float)
        m = mean_scores[m_name].to_numpy(dtype=float)
        y = mean_scores[y_name].to_numpy(dtype=float)
        a_coef, r2_m = self._standardized_ols(m, [x])
        y_coef, r2_y = self._standardized_ols(y, [x, m])
        a = float(a_coef[0])
        direct = float(y_coef[0])
        b = float(y_coef[1])
        indirect = a * b

        rng = np.random.default_rng(20260809)
        indirect_boot = []
        direct_boot = []
        n = len(mean_scores)
        for _ in range(int(bootstrap_iterations)):
            idx = rng.integers(0, n, n)
            try:
                a_b, _ = self._standardized_ols(m[idx], [x[idx]])
                y_b, _ = self._standardized_ols(y[idx], [x[idx], m[idx]])
                indirect_boot.append(float(a_b[0] * y_b[1]))
                direct_boot.append(float(y_b[0]))
            except Exception:
                continue
        if len(indirect_boot) < 100:
            raise RuntimeError("quantitative_bootstrap_insufficient")
        indirect_ci = np.percentile(indirect_boot, [2.5, 97.5])
        direct_ci = np.percentile(direct_boot, [2.5, 97.5])

        path_table = pd.DataFrame(
            [
                {"Association": f"{x_name} → {m_name}", "Standardized estimate": round(a, 3), "R² of outcome": round(r2_m, 3)},
                {"Association": f"{m_name} → {y_name} (controlling {x_name})", "Standardized estimate": round(b, 3), "R² of outcome": round(r2_y, 3)},
                {"Association": f"{x_name} → {y_name} (controlling {m_name})", "Standardized estimate": round(direct, 3), "R² of outcome": round(r2_y, 3)},
                {"Association": f"Indirect association: {x_name} → {m_name} → {y_name}", "Standardized estimate": round(indirect, 3), "R² of outcome": "—"},
            ]
        )
        bootstrap_table = pd.DataFrame(
            [
                {"Estimate": "Indirect association", "Point estimate": round(indirect, 3), "95% bootstrap CI lower": round(float(indirect_ci[0]), 3), "95% bootstrap CI upper": round(float(indirect_ci[1]), 3)},
                {"Estimate": "Residual direct association", "Point estimate": round(direct, 3), "95% bootstrap CI lower": round(float(direct_ci[0]), 3), "95% bootstrap CI upper": round(float(direct_ci[1]), 3)},
            ]
        )
        x_m_corr = float(np.corrcoef(x, m)[0, 1])
        vif = 1.0 / max(1e-9, 1.0 - x_m_corr**2)
        spearman = mean_scores.corr(method="spearman")
        summary = {
            "predictor": x_name,
            "mediator": m_name,
            "outcome": y_name,
            "a": round(a, 3),
            "b": round(b, 3),
            "direct": round(direct, 3),
            "indirect": round(indirect, 3),
            "r2_mediator": round(r2_m, 3),
            "r2_outcome": round(r2_y, 3),
            "indirect_ci": [round(float(indirect_ci[0]), 3), round(float(indirect_ci[1]), 3)],
            "direct_ci": [round(float(direct_ci[0]), 3), round(float(direct_ci[1]), 3)],
            "predictor_mediator_vif": round(float(vif), 3),
            "bootstrap_iterations": int(bootstrap_iterations),
        }
        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 ""
                    # Keep both ordinary decimals (0.31) and leading-decimal
                    # statistical notation (p < .05) inside one sentence.
                    leading_decimal = bool(
                        nxt.isdigit()
                        and re.search(r"(?:[<>=≤≥]|\b(?:p|r|β))\s*$", value[max(0, index - 12):index], flags=re.I)
                    )
                    decimal = (prev.isdigit() and nxt.isdigit()) or leading_decimal
                    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

    @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
        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 "")
        sample_size = meta.get("sample_size")
        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 (
                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))

            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 {
            "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,
            "uploaded_metric_anchors": uploaded_metric_anchors,
            "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 [],
        }

    @staticmethod
    def _semantic_frame_hash(payload):
        canonical = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

    @classmethod
    def build_quantitative_semantic_claim_frame(cls, sentence, metadata=None, section_name=""):
        """Freeze a compact server-side semantic frame for one quantitative claim.

        R6-R1 does not ask the model to decide this frame.  It records the
        source-side claim role, causal polarity/strength, methodological domain,
        hypothesis identity, and explicit result direction so downstream
        validation/translation can inherit one stable semantic authority.
        """
        text = re.sub(r"\s+", " ", str(sentence or "")).strip()
        folded = text.casefold()
        meta = dict(metadata or {})
        ledger = dict(meta.get("quantitative_evidence_ledger") or {})
        if not ledger and meta:
            try:
                ledger = cls.build_quantitative_evidence_ledger(meta)
            except Exception:
                ledger = {}
        constructs = dict(ledger.get("constructs") or {})
        known_constructs = [str(value) for value in constructs.values() if str(value)]
        ownership = cls._quantitative_sentence_ownership(text, section_name, known_constructs) if text else "other"

        relation_specs = (
            (r"\b(?:causes?|caused|causing)\b", "CAUSE"),
            (r"\b(?:leads?|led)\s+to\b", "LEAD_TO"),
            (r"\b(?:drives?|drove|driven)\b", "DRIVE"),
            (r"\b(?:determines?|determined)\b", "DETERMINE"),
            (r"\b(?:produces?|produced)\b", "PRODUCE"),
            (r"\b(?:results?|resulted)\s+in\b", "RESULT_IN"),
            (r"\b(?:yields?|yielded)\b", "YIELD"),
            (r"\b(?:generates?|generated)\b", "GENERATE"),
            (r"\bcausal(?:ity|\s+(?:effect|effects|mediation|mechanism|pathway|relationship|relationships|identification))\b", "CAUSAL_NOUN"),
        )
        relation = "NONE"
        for pattern, label in relation_specs:
            if re.search(pattern, folded, flags=re.I):
                relation = label
                break

        methodological_bias = cls._is_methodological_bias_mechanism_sentence(text, known_constructs)
        robustness_scope = bool(re.search(
            r"\b(?:robustness|sensitivity|spearman|rank[- ]order|extreme\s+values?|outliers?|influential\s+(?:cases|observations)|"
            r"response\s+styles?|common\s+method|same[- ]source|single\s+questionnaire|measurement\s+(?:error|bias))\b",
            folded, flags=re.I,
        ))
        if methodological_bias:
            domain = "METHODOLOGICAL_BIAS"
        elif robustness_scope:
            domain = "METHODOLOGICAL_ROBUSTNESS"
        elif re.search(r"\b(?:analysis|analyses|bootstrap|regression|correlation|estimation|statistical\s+(?:test|model|analysis)|vif)\b", folded):
            domain = "ANALYTICAL_REPORTING"
        elif any(name.casefold() in folded for name in known_constructs):
            domain = "SUBSTANTIVE_CONSTRUCT"
        else:
            domain = "GENERAL"

        raw_causal_state = cls.classify_quantitative_causal_sentence(text) if text else "NON_CAUSAL"
        validation_state = "NON_CAUSAL" if methodological_bias else raw_causal_state
        if raw_causal_state == "EXPLICIT_CAUSAL_LIMITATION":
            polarity = "NEGATIVE"
            causal_strength = "CAUSAL_LIMITATION"
        elif raw_causal_state == "AFFIRMATIVE_CAUSAL":
            polarity = "AFFIRMATIVE"
            causal_strength = "CAUSAL_ASSERTION"
        elif re.search(r"\b(?:association|associations|associated|correlation|correlations|correlated|covariation|covariance)\b", folded):
            polarity = "NONE"
            causal_strength = "ASSOCIATIONAL"
        else:
            polarity = "NONE"
            causal_strength = "NON_CAUSAL"

        if re.search(r"\b(?:may|might|could|appears?\s+to|seems?\s+to|suggests?|consistent\s+with)\b", folded) and not re.search(r"\bcould\s+not\b", folded):
            modality = "HEDGED"
        elif raw_causal_state == "EXPLICIT_CAUSAL_LIMITATION":
            modality = "LIMITATION"
        else:
            modality = "ASSERTED"

        hypothesis_ids = sorted(set(re.findall(r"(?i)\bH\d+[A-Za-z]?\b", text)), key=str.casefold)
        if re.search(r"\b(?:non[- ]?significant|not\s+significant|did\s+not\s+support|was\s+not\s+supported)\b", folded):
            result_direction = "NOT_SUPPORTED_OR_NONSIGNIFICANT"
        elif re.search(r"\b(?:positive|positively)\b[^.;]{0,80}\b(?:association|associated|relationship|correlation|estimate|coefficient)\b", folded):
            result_direction = "POSITIVE"
        elif re.search(r"\b(?:negative|negatively)\b[^.;]{0,80}\b(?:association|associated|relationship|correlation|estimate|coefficient)\b", folded):
            result_direction = "NEGATIVE"
        else:
            result_direction = "UNSPECIFIED"

        if hypothesis_ids:
            claim_type = "HYPOTHESIS_CLAIM"
        elif domain in {"METHODOLOGICAL_BIAS", "METHODOLOGICAL_ROBUSTNESS"}:
            claim_type = "METHODOLOGICAL_CLAIM"
        elif ownership == "result":
            claim_type = "RESULT_CLAIM"
        elif causal_strength == "ASSOCIATIONAL":
            claim_type = "ASSOCIATION_CLAIM"
        elif raw_causal_state != "NON_CAUSAL":
            claim_type = "CAUSAL_SCOPE_CLAIM"
        else:
            claim_type = "OTHER"

        frame = {
            "version": cls.SEMANTIC_CLAIM_FRAME_VERSION,
            "mode": "quantitative",
            "claim_type": claim_type,
            "ownership": ownership,
            "domain": domain,
            "relation": relation,
            "polarity": polarity,
            "causal_strength": causal_strength,
            "causal_state": raw_causal_state,
            "validation_state": validation_state,
            "modality": modality,
            "hypothesis_ids": hypothesis_ids,
            "result_direction": result_direction,
        }
        frame["frame_hash"] = cls._semantic_frame_hash(frame)
        return frame

    @classmethod
    def _classify_quantitative_causal_clause(cls, clause):
        """Classify one clause by causal *assertion scope*, not token presence.

        C5.5 freezes a single semantic contract for Writer/Validator/Retry. A
        causal phrase inside an explicit limitation (for example, ``cannot
        establish that X causes Y``) is not an affirmative claim. Conversely,
        an actual causal assertion remains fail-closed even when another part
        of the sentence contains a disclaimer.
        """
        value = re.sub(r"\s+", " ", str(clause or "")).strip().casefold()
        if not value:
            return "NON_CAUSAL"

        causal_term = (
            r"(?:causal(?:ity|\s+(?:effect|effects|mediation|mechanism|pathway|relationship|relationships|identification))|"
            r"causation|cause(?:s|d|ing)?)"
        )
        direct_patterns = (
            (r"\b(?:causes?|caused|causing)\b", "STRONG"),
            (r"\b(?:leads?|led)\s+to\b", "STRONG"),
            (r"\b(?:drives?|drove|driven)\b", "STRONG"),
            (r"\b(?:determines?|determined)\b", "STRONG"),
            (r"\b(?:produces?|produced)\b", "AMBIGUOUS_REPORTING"),
            (r"\b(?:results?|resulted)\s+in\b", "AMBIGUOUS_REPORTING"),
            (r"\b(?:yields?|yielded)\b", "AMBIGUOUS_REPORTING"),
            (r"\b(?:generates?|generated)\b", "AMBIGUOUS_REPORTING"),
            (r"\b(?:transmits?|transmitted)\s+(?:an?\s+|the\s+)?causal\s+effect\b", "STRONG"),
            (r"\bcausally\s+(?:transmits?|mediates?|affects?|influences?)\b", "STRONG"),
            (r"\b(?:has|have|had|exerts?|exerted)\s+(?:an?\s+|the\s+)?causal\s+effect\b", "STRONG"),
        )

        def local_actor_text(match):
            """Return the grammatical actor window immediately before a predicate.

            R3 deliberately classifies the *actor* rather than maintaining an
            ever-growing whitelist of reporting artefacts.  Leading adverbial or
            prepositional phrases (``In the analysis, ...`` / ``Using Spearman
            correlations, ...``) are peeled away so that a nearby mention of an
            analysis cannot shield a later construct-level causal assertion.
            """
            prefix = value[max(0, match.start() - 260):match.start()]
            actor = re.split(r"[.;!?]", prefix)[-1].strip()
            lead = re.match(
                r"^(?:in|using|with|based\s+on|according\s+to|after|before|when|while|because|"
                r"although|though|given|under|for|as)\b[^,]{0,180},\s*(.+)$",
                actor,
                flags=re.I,
            )
            if lead:
                actor = lead.group(1).strip()
            return actor[-220:]

        def actor_type(match):
            """Classify the predicate subject as analytical, weak-analytical, or other.

            ``ANALYTICAL_ACTOR`` is intentionally restricted to explicit
            statistical/diagnostic actors.  Generic ``model``/``procedure`` terms
            are weak because they can also denote the substantive AI system or an
            intervention; they are exempt only when the predicate target is a
            recognisable reporting artefact.
            """
            actor = local_actor_text(match)
            strong_actor = (
                r"\b(?:analysis|analyses|"
                r"bootstrap(?:\s+(?:resampling|procedure|computation|analysis))?|"
                r"resampling(?:\s+(?:procedure|analysis|computation))?|"
                r"regression\s+(?:analysis|model)|association\s+(?:analysis|model)|"
                r"mediation\s+(?:analysis|model)|"
                r"estimation(?:\s+(?:procedure|method|analysis))?|estimator|"
                r"statistical\s+(?:analysis|model|procedure|method|test)|"
                r"robustness\s+(?:check|test|analysis)|"
                r"sensitivity\s+(?:check|test|analysis)|"
                r"diagnostic\s+(?:check|test|analysis)|"
                r"rank[- ]order\s+(?:robustness\s+)?check|"
                r"(?:spearman|pearson|rank[- ]order)\s+correlations?|"
                r"correlation\s+(?:analysis|matrix|check|test)|"
                r"hypothesis\s+test|significance\s+test|"
                r"variance\s+inflation\s+factor(?:\s+check)?|vif(?:\s+check)?|"
                r"measurement\s+assessment|descriptive\s+analysis)\b"
            )
            if re.search(strong_actor, actor, flags=re.I):
                return "ANALYTICAL_ACTOR"
            if re.search(r"\b(?:model|procedure|method|computation|calculation)\b", actor, flags=re.I):
                return "WEAK_ANALYTICAL_ACTOR"
            return "OTHER_ACTOR"

        def predicate_target_is_reporting_output(match):
            suffix = value[match.end():min(len(value), match.end() + 220)]
            target_pattern = (
                r"^\s*(?:(?:an?|the|this|that|these|those|following|reported|"
                r"standardized|standardised|numerical|estimated|95\s*percent|95%)\s+){0,6}"
                r"(?:confidence\s+intervals?|interval\s+estimates?|"
                r"coefficient(?:\s+estimates?)?s?|standard\s+errors?|"
                r"parameter\s+estimates?|point\s+estimates?|indirect\s+estimates?|"
                r"direct\s+estimates?|effect-size\s+estimates?|"
                r"correlations?|covariances?|variances?|loadings?|"
                r"test\s+statistics?|p[- ]?values?|vif(?:s|\s+values)?|"
                r"cronbach(?:'s)?\s+alpha|composite\s+reliability|"
                r"average\s+variance\s+extracted|ave|numerical\s+estimates?|estimates?|"
                r"(?:a\s+)?pattern(?:\s+of\s+(?:associations|correlations|estimates))?|"
                r"(?:a\s+)?(?:similar|consistent|comparable)\s+pattern|"
                r"findings?|results?)\b"
            )
            return bool(re.search(target_pattern, suffix, flags=re.I))

        def direct_match_is_analytical_reporting(match, predicate_kind):
            actor = actor_type(match)
            if predicate_kind == "AMBIGUOUS_REPORTING":
                # ``analysis/check/test produced/resulted/yielded/generated ...``
                # describes what an analytical operation reported.  It is not a
                # study-construct causal assertion even when the following prose is
                # stylistically loose (e.g. ``the analysis produced higher
                # engagement``).  The CAR layer may flag wording separately; this
                # causal gate must not terminate the manuscript for it.
                if actor == "ANALYTICAL_ACTOR":
                    return True
                if actor == "WEAK_ANALYTICAL_ACTOR" and predicate_target_is_reporting_output(match):
                    return True
                return False

            # Strong causal predicates remain fail-closed.  Preserve only the
            # narrow statistical-reporting idiom already accepted by R1, such as
            # ``the analysis led to the following numerical estimate``.
            return actor in {"ANALYTICAL_ACTOR", "WEAK_ANALYTICAL_ACTOR"} and predicate_target_is_reporting_output(match)

        def direct_match_is_disclaimed(match):
            prefix = value[max(0, match.start() - 220):match.start()]
            suffix = value[match.end():min(len(value), match.end() + 220)]

            # R6-R1 predicate polarity closure: passive negation is a semantic
            # limitation ("associations are not driven by outliers"), not an
            # affirmative causal claim.  Do not exempt contrastive constructions
            # that merely replace one asserted cause with another ("not driven by
            # X but rather by Y") or scalar forms such as "not only driven".
            passive_negative = bool(re.search(
                r"\b(?:is|are|was|were|be|been|being)\s+"
                r"(?:(?:clearly|necessarily|directly|statistically)\s+){0,2}"
                r"(?:not|never)\s+"
                r"(?:(?:clearly|necessarily|directly|statistically)\s+){0,2}$",
                prefix, flags=re.I,
            ))
            if passive_negative:
                alternate_cause = re.search(
                    r"\b(?:but|rather|instead)\b\s*(?:rather\s+|instead\s+|also\s+)?by\b",
                    suffix, flags=re.I,
                )
                alternate_negated = re.search(
                    r"\b(?:but|rather|instead)\b\s*(?:also\s+)?(?:not|never)\s+by\b",
                    suffix, flags=re.I,
                )
                if alternate_cause and not alternate_negated:
                    return False
                return True

            if re.search(
                r"(?:\b(?:does|do|did|can|could|would|should|must|may|might)\s+not|"
                r"\bcannot|\bcan't|\bcouldn't|\bnever)\s+(?:\w+\s+){0,2}$",
                prefix,
                flags=re.I,
            ):
                return True
            if re.search(
                r"\b(?:cannot|can't|could not|couldn't|does not|doesn't|do not|don't|did not|didn't|"
                r"fails? to|failed to|unable to|insufficient to|not able to)\s+"
                r"(?:identify|establish|demonstrate|show|infer|support|imply|mean|claim|justify|warrant|prove)"
                r"\b[^.;]{0,180}?(?:\bthat\b|\bwhether\b)[^.;]{0,120}$",
                prefix,
                flags=re.I,
            ):
                return True
            if re.search(
                r"\b(?:should not|shouldn't|must not|cannot|can't|could not|couldn't)\s+be\s+"
                r"(?:interpreted|read|treated|understood)\s+(?:as|to mean)\b[^.;]{0,140}$",
                prefix,
                flags=re.I,
            ):
                return True
            if re.search(
                r"\bwithout\s+(?:identifying|establishing|demonstrating|showing|inferring|supporting|proving)"
                r"\b[^.;]{0,180}?(?:\bthat\b|\bwhether\b)[^.;]{0,120}$",
                prefix,
                flags=re.I,
            ):
                return True
            return False

        saw_disclaimed_direct = False
        for pattern, predicate_kind in direct_patterns:
            for match in re.finditer(pattern, value, flags=re.I):
                if direct_match_is_disclaimed(match):
                    saw_disclaimed_direct = True
                    continue
                if direct_match_is_analytical_reporting(match, predicate_kind):
                    continue
                return "AFFIRMATIVE_CAUSAL"

        affirmative_noun_patterns = (
            rf"\b(?:is|are|was|were|represent(?:s|ed)?|constitut(?:es|ed))\s+(?:an?\s+|the\s+)?{causal_term}\b",
            rf"\b(?:identify|identifies|identified|establish|establishes|established|demonstrate|demonstrates|demonstrated|show|shows|showed|prove|proves|proved|provide|provides|provided)\b[^.;]{{0,80}}\b{causal_term}\b",
            rf"\b(?:evidence|proof)\s+(?:of|for)\s+(?:an?\s+|the\s+)?{causal_term}\b",
            rf"\b{causal_term}\b[^.;]{{0,80}}\b(?:is|are|was|were)\s+(?!not\b|never\b)",
        )
        for pattern in affirmative_noun_patterns:
            for match in re.finditer(pattern, value, flags=re.I):
                prefix = value[max(0, match.start() - 80):match.start()]
                if re.search(
                    r"\b(?:no|not|without|cannot|can't|insufficient|unable|does not|doesn't|do not|don't|did not|didn't|not able)\b",
                    prefix,
                    flags=re.I,
                ):
                    continue
                return "AFFIRMATIVE_CAUSAL"

        limitation_patterns = (
            rf"\b(?:cannot|can't|could not|couldn't|does not|doesn't|do not|don't|did not|didn't|fails? to|failed to|"
            rf"is not able to|are not able to|was not able to|were not able to|unable to|insufficient to|not able to)\s+"
            rf"(?:identify|establish|demonstrate|show|infer|support|imply|mean|claim|justify|warrant|permit|allow|prove)\b[^.;]*\b{causal_term}\b",
            rf"\b(?:is|are|was|were)\s+not\s+(?:designed|intended)\s+to\s+"
            rf"(?:identify|establish|demonstrate|show|infer|support|prove)\b[^.;]*\b{causal_term}\b",
            rf"\bwithout\s+(?:identifying|establishing|demonstrating|showing|inferring|supporting|proving)\b[^.;]*\b{causal_term}\b",
            rf"\b(?:no|without)\s+(?:evidence|basis|support|identification|proof)\b[^.;]*\b{causal_term}\b",
            rf"\b(?:is|are|was|were)\s+not\s+(?:evidence|proof)\s+(?:of|for)\s+(?:an?\s+|the\s+)?{causal_term}\b",
            rf"\b(?:cannot|can't|could not|couldn't)\s+be\s+used\s+to\s+(?:infer|establish|demonstrate|show|prove)\b[^.;]*\b{causal_term}\b",
            rf"\b(?:should not|shouldn't|must not|cannot|can't|could not|couldn't)\s+be\s+"
            rf"(?:interpreted|read|treated|understood)\s+(?:as|to mean)\b[^.;]*\b{causal_term}\b",
            rf"\b(?:is|are|was|were)?\s*(?:restricted|limited|confined)\s+to\b[^.;]*"
            rf"\b(?:not|rather than|instead of)\s+(?:to\s+)?(?:an?\s+|the\s+)?{causal_term}\b",
            rf"\b(?:association|associations|covariation|covariance|correlation|correlations|indirect association|descriptive pattern|descriptive patterns)\b[^.;]*"
            rf"\b(?:not(?:\s+as)?|rather than|instead of)\s+(?:an?\s+|the\s+)?{causal_term}\b",
            rf"\b{causal_term}\b\s+(?:is|are|was|were)\s+(?:not|never)\s+"
            rf"(?:identified|established|demonstrated|shown|inferred|supported|claimed|justified|warranted|proven)\b",
            rf"\b(?:not|no)\s+(?:an?\s+|the\s+)?{causal_term}\b",
        )
        if saw_disclaimed_direct or any(re.search(pattern, value, flags=re.I) for pattern in limitation_patterns):
            return "EXPLICIT_CAUSAL_LIMITATION"
        return "NON_CAUSAL"

    @classmethod
    def classify_quantitative_causal_sentence(cls, sentence):
        """Three-state causal contract shared by inspection, repair, and tests."""
        value = str(sentence or "")
        folded = re.sub(r"\s+", " ", value).strip().casefold()
        # A negated passive can still assert an alternative cause: "not driven
        # by X but rather by Y".  Detect this before contrast-clause splitting,
        # otherwise the second fragment loses the causal predicate surface.
        alternate_cause = re.search(
            r"\b(?:is|are|was|were|be|been|being)\s+"
            r"(?:(?:clearly|necessarily|directly|statistically)\s+){0,2}"
            r"(?:not|never)\s+"
            r"(?:(?:clearly|necessarily|directly|statistically)\s+){0,2}"
            r"(?:caused|driven|determined|produced|generated|yielded)\b"
            r"[^.;]{0,180}\b(?:but|rather|instead)\b\s*"
            r"(?:rather\s+|instead\s+|also\s+)?by\b",
            folded, flags=re.I,
        )
        if alternate_cause:
            return "AFFIRMATIVE_CAUSAL"
        # Contrast clauses must not let an early disclaimer shield a later assertion.
        clauses = re.split(
            r"(?i)(?:;|\bbut\b|\bhowever\b|\bwhereas\b|\byet\b|\bnevertheless\b|\balthough\b|\bthough\b)",
            value,
        )
        states = [cls._classify_quantitative_causal_clause(part) for part in clauses if str(part).strip()]
        if "AFFIRMATIVE_CAUSAL" in states:
            return "AFFIRMATIVE_CAUSAL"
        if "EXPLICIT_CAUSAL_LIMITATION" in states:
            return "EXPLICIT_CAUSAL_LIMITATION"
        return "NON_CAUSAL"

    @classmethod
    def _causal_phrase_is_explicitly_disclaimed(cls, sentence, match_start=None, match_end=None):
        """Compatibility wrapper for older call sites; uses the shared tri-state contract."""
        return cls.classify_quantitative_causal_sentence(sentence) == "EXPLICIT_CAUSAL_LIMITATION"

    @classmethod
    def _is_methodological_bias_mechanism_sentence(cls, sentence, known_constructs=None):
        """Recognize measurement/method-bias mechanisms without weakening construct causality.

        R4 separates methodological statements about how measurement conditions
        may bias statistical quantities from substantive claims that one study
        construct causes another.  The exemption is intentionally narrow: a
        measurement/source-bias actor and a statistical/measurement target must
        both be present.  Explicit construct-level causal language still wins.
        """
        value = re.sub(r"\s+", " ", str(sentence or "")).strip().casefold()
        if not value:
            return False
        methodological_actor = re.search(
            r"\b(?:self[- ]reports?|single\s+questionnaire|same[- ]source|common\s+method(?:\s+bias)?|"
            r"response\s+styles?|social\s+desirability|measurement\s+error|item\s+wording|"
            r"common\s+rater|method\s+variance|single[- ]source\s+measurement)\b", value, re.I
        )
        statistical_target = re.search(
            r"\b(?:correlations?|covariances?|associations?|estimates?|coefficients?|variance|"
            r"standard\s+errors?|reliability|validity|measurement\s+error|method\s+variance)\b", value, re.I
        )
        mechanism = re.search(
            r"\b(?:inflate|inflates|inflated|attenuate|attenuates|attenuated|bias|biases|biased|"
            r"distort|distorts|distorted|increase|increases|increased|decrease|decreases|decreased|"
            r"contribute|contributes|contributed|induce|induces|induced)\b", value, re.I
        )
        if not (methodological_actor and statistical_target and mechanism):
            return False

        # Never let a methodological phrase shield an explicit relation between
        # named study constructs.
        constructs = [str(x or '').strip().casefold() for x in (known_constructs or []) if str(x or '').strip()]
        named_hits = [c for c in constructs if c and c in value]
        if len(named_hits) >= 2 and re.search(
            r"\b(?:causes?|leads?\s+to|drives?|determines?|produces?|results?\s+in|yields?|generates?)\b",
            value, re.I
        ):
            return False
        return True

    @classmethod
    def inspect_quantitative_study_fact_contract(cls, text, metadata, section_name=""):
        """Pure AMJ quantitative Evidence Ledger inspection; never rewrites prose."""
        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)
        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 {})
        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)",
        }
        violations = []

        def add_metric_violations(sentence, pattern, expected, *, dtype="float", label="metric"):
            if expected is None:
                return
            regex = re.compile(pattern, re.I)
            for match in regex.finditer(sentence):
                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 not same:
                    expected_text = str(int(expected)) if dtype == "int" else f"{float(expected):.3f}".rstrip("0").rstrip(".")
                    violations.append({
                        "type":"quantitative_metric_mismatch", "metric":label,
                        "observed":observed, "expected":expected_text,
                        "preview":str(sentence).strip()[:180],
                    })

        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
            if sample_size:
                for match in re.finditer(r"(?i)(\b(?:sample(?:\s+size)?(?:\s+of)?|N)\s*(?:=|:)?\s*)(\d{2,6})\b", sentence):
                    observed = int(match.group(2))
                    if observed != int(sample_size):
                        violations.append({"type":"sample_size_mismatch","observed":observed,"expected":int(sample_size),"preview":sentence.strip()[:180]})
                for match in re.finditer(r"(?i)\b(\d{2,6})\s+(respondents|participants|observations|cases)\b", sentence):
                    observed = int(match.group(1))
                    if observed != int(sample_size):
                        violations.append({"type":"sample_count_mismatch","observed":observed,"expected":int(sample_size),"preview":sentence.strip()[:180]})
            if scale_points:
                for match in re.finditer(r"(?i)\b([2-9]|1[0-1])[- ]point\s+(Likert(?:-type)?\s+scale)", sentence):
                    observed = int(match.group(1))
                    if observed != int(scale_points):
                        violations.append({"type":"scale_points_mismatch","observed":observed,"expected":int(scale_points),"preview":sentence.strip()[:180]})
            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(sentence):
                        violations.append({"type":"construct_role_mismatch","role":role,"observed":candidate,"expected":expected,"preview":sentence.strip()[:180]})
            if strict:
                methodological_bias_scope = cls._is_methodological_bias_mechanism_sentence(
                    sentence, known_constructs
                )
                semantic_frame = cls.build_quantitative_semantic_claim_frame(
                    sentence, meta, section_name
                )
                causal_state = str(semantic_frame.get("validation_state") or "NON_CAUSAL")
                if causal_state == "AFFIRMATIVE_CAUSAL":
                    violations.append({
                        "type":"causal_language_violation",
                        "preview":sentence.strip()[:180],
                        "reason":"The sentence makes an affirmative causal claim that exceeds the cross-sectional observational evidence contract.",
                        "repair_instruction":(
                            "Preserve the study meaning but rewrite as association/covariation only. "
                            "Explicit causal-limit language is allowed, including forms such as "
                            "'restricted to associations, not causal effects' or 'cannot establish causality'."
                        ),
                    })
                if (any(c in sentence.casefold() for c in construct_folded) if construct_folded else False) and re.search(r"(?i)\bp\s*(?:=|<|>|≤|≥)\s*(?:0?\.\d+|\d+(?:\.\d+)?)", sentence):
                    violations.append({"type":"unfrozen_p_value","preview":sentence.strip()[:180]})
                if ownership == "study":
                    bad = next((term for term in forbidden if re.search(rf"(?i)(?<![A-Za-z]){re.escape(term)}(?![A-Za-z])", sentence)), "")
                    if bad:
                        violations.append({"type":"unsupported_quantitative_method","method":bad,"preview":sentence.strip()[:180]})

            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")
            add_metric_violations(sentence, r"(\bVIF\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", vif_expected, label="VIF")
            add_metric_violations(sentence, r"(\bbootstrap\s+(?:resamples?|iterations?)\s*(?:=|:)?\s*)(\d{2,6})", boot_expected, 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:
                        add_metric_violations(sentence, rf"(R(?:²|\^2)\s*(?:for|of)?\s*{re.escape(construct)}\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", results.get(key), label=key)
                lower = sentence.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")]),
                ]
                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:
                    add_metric_violations(sentence, r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", selected[1], label=selected[0])
            elif not strict:
                for label, expected in dict(uploaded.get("r2_by_label") or {}).items():
                    if label and label in sentence.casefold():
                        add_metric_violations(sentence, rf"(R(?:²|\^2)\s*(?:for|of)\s*{re.escape(label)}\s*(?:=|:)\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_R2:{label}")
                for path, expected in dict(uploaded.get("beta_by_path") or {}).items():
                    left, _, right = path.partition(" -> ")
                    if left and right and left in sentence.casefold() and right in sentence.casefold():
                        add_metric_violations(sentence, r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_beta:{path}")
        return violations

    @classmethod
    def repair_quantitative_study_fact_contract(cls, text, metadata, section_name=""):
        """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 {})
        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 {})
        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"),
        )

        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):
                    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)

            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)}")
                    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:
                causal_state = cls.classify_quantitative_causal_sentence(revised)
                if causal_state == "AFFIRMATIVE_CAUSAL":
                    for pattern, replacement in causal_replacements:
                        regex = re.compile(pattern)
                        revised, repaired_count = regex.subn(replacement, revised)
                        if repaired_count:
                            diagnostics.append({
                                "type":"causal_language_repaired",
                                "replacement":replacement,
                                "count":repaired_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 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:
                        diagnostics.append({"type":"unsupported_quantitative_method_removed","method":bad,"preview":sentence.strip()[:180]})
                        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(".")
                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
                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 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_local_metric(metric_pattern, results.get(key), label=key)

                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")]),
                ]
                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)
        return value.strip(), diagnostics

    @classmethod
    def validate_quantitative_study_fact_contract(cls, text, metadata, section_name=""):
        violations = cls.inspect_quantitative_study_fact_contract(text, metadata, section_name=section_name)
        if violations:
            raise RuntimeError(f"quantitative_study_fact_contract_invalid:{section_name}:{violations[:3]}")
        return True

    def resolve_empirical_data(
        self,
        uploaded_files=None,
        language="zh",
        *,
        topic="",
        keywords="",
        agent=None,
    ):
        if uploaded_files:
            user_md = extract_data_from_files(uploaded_files)
            if user_md:
                evidence_sections = [
                    heading.strip()
                    for heading in re.findall(
                        r"(?m)^#{1,3}\s+(.+?)\s*$",
                        user_md,
                    )
                    if heading.strip()
                ]
                metadata = {
                    "mode": "quantitative",
                    "source_type": "user_provided",
                    "disclosure_required": False,
                    "evidence_sections": evidence_sections[:20]
                    or ["User-provided quantitative evidence"],
                    "evidence_boundary": (
                        "Only values explicitly present in the uploaded evidence are frozen. "
                        "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)
        construct_names = design["constructs"]
        df, validity_df, fl_matrix_df, mean_scores, corr_matrix = self.generate_empirical_data_robust(
            construct_names=construct_names
        )
        path_summary, path_table, spearman = self._path_evidence(mean_scores, construct_names)
        context = design["research_context"]
        generated_prompt = f"""### Frozen Quantitative Study Design
| Field | Frozen value |
| :--- | :--- |
| Design | Cross-sectional, single-wave questionnaire study |
| Respondent context | {context} |
| Sample size | {len(df)} |
| Response scale | 7-point Likert-type scale |
| Predictor | {construct_names[0]} |
| Mediator | {construct_names[1]} |
| Outcome | {construct_names[2]} |
| Causal identification | None; report associations and estimated indirect association, not causal effects |

### Table 1: Reliability and Convergent Validity
{validity_df.to_markdown(index=False)}

### Table 2: Descriptive Statistics and Discriminant Validity (Fornell-Larcker Criterion)
*Note: Diagonal elements in parentheses are the square roots of AVE.*
{fl_matrix_df.to_markdown()}

### Table 3: Frozen Association Model Results
{path_table.to_markdown(index=False)}

### Table 4: Rank-Order Robustness Correlations (Spearman)
{spearman.round(3).to_markdown()}

### Frozen Bootstrap and Collinearity Checks
- Bootstrap resamples: {path_summary['bootstrap_iterations']}.
- Indirect association ({construct_names[0]} → {construct_names[1]} → {construct_names[2]}): {path_summary['indirect']}; 95% bootstrap CI [{path_summary['indirect_ci'][0]}, {path_summary['indirect_ci'][1]}].
- Residual direct association ({construct_names[0]} → {construct_names[2]} controlling {construct_names[1]}): {path_summary['direct']}; 95% bootstrap CI [{path_summary['direct_ci'][0]}, {path_summary['direct_ci'][1]}].
- Predictor/mediator VIF in the outcome model: {path_summary['predictor_mediator_vif']}.

[STRICT EMPIRICAL INSTRUCTION]
1. ZERO HALLUCINATION: Every study-specific fact and number must come from this frozen packet. Never invent a country, named organization, sampling frame, response rate, demographic characteristic, measure source, software package, coefficient, p-value, fit index, robustness test, or date.
2. DESIGN DISCIPLINE: This is a cross-sectional observational survey. Use association language. The indirect estimate is not causal mediation identification.
3. MEASUREMENT DEFENSE: State reliability/convergent/discriminant validity only when the displayed values support it.
4. RESULTS DISCIPLINE: Results must report this study's frozen numbers without attaching external citations to the coefficients or sample facts.
5. ROBUSTNESS: The Spearman matrix, bootstrap intervals, and VIF above are the only frozen robustness evidence. Do not add another robustness procedure.
6. CONSISTENCY: The predictor, mediator, outcome, sample size, context, scale, and all numerical results above are the single source of quantitative truth throughout the manuscript.
"""
        metadata = {
            "mode": "quantitative",
            "source_type": "system_generated",
            "disclosure_required": True,
            "sample_size": int(len(df)),
            "scale_points": 7,
            "study_design": "cross-sectional single-wave questionnaire",
            "research_context": context,
            "constructs": {
                "predictor": construct_names[0],
                "mediator": construct_names[1],
                "outcome": construct_names[2],
            },
            "association_results": path_summary,
            "method": (
                "Positive-semidefinite latent correlation generation, indicator loadings, "
                "Likert discretization, reliability/validity summaries, standardized association "
                "models, fixed-seed bootstrap intervals, and Spearman robustness correlations."
            ),
            "generation_parameters": {
                "constructs": construct_names,
                "sample_size": int(len(df)),
                "likert_scale_points": 7,
                "items_per_construct": "random integer from 3 to 5",
                "indicator_loading_range": [0.72, 0.88],
                "latent_correlation_control": "positive-semidefinite matrix with maximum target correlation 0.55",
                "item_generation": "loading × latent score + normally distributed measurement error",
                "discretization": "round to nearest integer and clamp to 1–7",
                "bootstrap_iterations": path_summary["bootstrap_iterations"],
                "bootstrap_seed": 20260809,
                "reported_summaries": [
                    "Cronbach alpha",
                    "composite reliability",
                    "average variance extracted",
                    "descriptive statistics",
                    "Fornell-Larcker matrix",
                    "standardized association estimates",
                    "R-squared",
                    "bootstrap confidence intervals",
                    "Spearman correlations",
                    "VIF",
                ],
            },
            "evidence_sections": [
                "Frozen quantitative study design",
                "Reliability and convergent validity",
                "Descriptive statistics and discriminant validity",
                "Frozen association model results",
                "Rank-order robustness correlations",
                "Frozen bootstrap and collinearity checks",
            ],
        }
        metadata["semantic_claim_frame_version"] = self.SEMANTIC_CLAIM_FRAME_VERSION
        metadata["hypothesis_ledger"] = self.build_hypothesis_ledger(construct_names)
        metadata["quantitative_table_manifest"] = self.build_table_manifest(
            validity_df, fl_matrix_df, path_table, spearman
        )
        metadata["quantitative_evidence_ledger"] = self.build_quantitative_evidence_ledger(metadata, generated_prompt)
        return (
            df,
            generated_prompt,
            tr(language, "synthetic_stats_generated"),
            metadata,
        )


def get_quantitative_engine():
    return QuantitativeEngine()