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
from core.study_design_contract import (
    STUDY_DESIGN_CONTRACT_VERSION,
    build_quantitative_contract,
    validate_quantitative_contract,
    compile_quantitative_hypothesis_ledger,
    build_quantitative_front_matter_contract,
    build_user_research_intent,
)


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"
    FRONT_MATTER_CONTRACT_VERSION = "quantitative-front-matter-contract-v2"
    TARGET_TABLE_BINDING_VERSION = "target-quantitative-table-binding-v1"
    SEMANTIC_CLAIM_FRAME_VERSION = "smartpaper-core-semantic-claim-frame-v1"
    STUDY_DESIGN_CONTRACT_VERSION = STUDY_DESIGN_CONTRACT_VERSION

    _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_front_matter_contract(cls, topic, keywords, constructs):
        """Build a reader-front-matter semantic contract from adaptive design authority.

        ``constructs`` accepts either the new Study Design Contract or the legacy
        three-construct list/dict so old checkpoints remain readable.
        """
        if isinstance(constructs, dict) and constructs.get("version") == cls.STUDY_DESIGN_CONTRACT_VERSION:
            return build_quantitative_front_matter_contract(constructs, version=cls.FRONT_MATTER_CONTRACT_VERSION)
        if isinstance(constructs, dict) and "quantitative" in constructs:
            return build_quantitative_front_matter_contract(constructs, version=cls.FRONT_MATTER_CONTRACT_VERSION)
        if isinstance(constructs, dict):
            values = [constructs.get(role) for role in ("predictor", "mediator", "outcome")]
        else:
            values = list(constructs or [])
        candidate = {"constructs": values, "research_context": "respondents with experience relevant to the focal management phenomenon"}
        contract = build_quantitative_contract(topic, keywords, candidate, source_type="system_generated")
        return build_quantitative_front_matter_contract(contract, version=cls.FRONT_MATTER_CONTRACT_VERSION)

    @classmethod
    def build_hypothesis_ledger(cls, constructs):
        """Compile H1-Hn from a frozen adaptive graph; preserve legacy H1-H3 order."""
        if isinstance(constructs, dict) and constructs.get("version") == cls.STUDY_DESIGN_CONTRACT_VERSION:
            contract = constructs
        elif isinstance(constructs, dict) and "quantitative" in constructs:
            contract = constructs
        else:
            values = list(constructs.values()) if isinstance(constructs, dict) else list(constructs or [])
            contract = build_quantitative_contract("", "", {"constructs": values}, source_type="system_generated")
        return compile_quantitative_hypothesis_ledger(contract, version=cls.HYPOTHESIS_LEDGER_VERSION)

    @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):
        """Legacy helper retained for tests/checkpoints; no longer defines model topology."""
        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[:8]

    @staticmethod
    def _contract_construct_labels(contract):
        return [
            str(item.get("label") or "").strip()
            for item in dict(contract or {}).get("quantitative", {}).get("constructs", [])
            if isinstance(item, dict) and str(item.get("label") or "").strip()
        ]

    def _derive_system_design(self, agent, topic="", keywords=""):
        """Interpret a parsimonious topic-aligned quantitative graph adaptively.

        The existing ``structured_json`` logical node is intentionally reused so
        Standard/Flash and Flagship/Pro ModelPolicy routing remains unchanged.
        """
        user_intent = build_user_research_intent(topic, keywords, mode="quantitative")
        fallback_candidate = {
            "constructs": [
                {"label":"Focal Predictor","roles":["predictor"]},
                {"label":"Focal Mediator","roles":["mediator"]},
                {"label":"Focal Outcome","roles":["outcome"]},
            ],
            "relations": [
                {"type":"direct_association","source":"Focal Predictor","target":"Focal Outcome","hypothesis_order":1},
                {"type":"direct_association","source":"Focal Predictor","target":"Focal Mediator","hypothesis_order":2},
                {"type":"direct_association","source":"Focal Mediator","target":"Focal Outcome","hypothesis_order":3},
            ],
            "indirect_effects":[{"path":["Focal Predictor","Focal Mediator","Focal Outcome"],"kind":"simple"}],
            "research_context":"respondents with experience relevant to the focal management phenomenon",
            "background_concepts":[],
        }
        if agent is None:
            return build_quantitative_contract(topic, keywords, fallback_candidate)
        prompt = f"""Design the minimum coherent quantitative management-study graph for the topic below.
Return ONLY valid JSON with this schema:
{{
  "constructs":[{{"label":"construct","roles":["predictor|mediator|moderator|outcome|control|covariate"]}}],
  "relations":[{{"type":"direct_association|moderation|control_effect","source":"construct label","target":"construct label","moderator":"construct label or null","hypothesis_order":1}}],
  "indirect_effects":[{{"path":["construct A","construct B","construct C"],"kind":"simple|serial|parallel"}}],
  "research_context":"one generic respondent context",
  "background_concepts":["topic concept that is contextual rather than measured/modelled"]
}}
Rules:
0. USER RESEARCH INTENT AUTHORITY: The user's topic and keywords are deliberate research-design inputs. Never delete, replace, or downgrade an explicitly requested focal construct, mediation, moderation, serial/parallel mediation, control, population, or research context merely because direct literature support may be sparse. Literature evidence constrains claims about PRIOR research; it does not redefine what this study is asked to investigate.
1. Use 2-8 distinct constructs only when the topic actually requires them; do NOT force a mediator or moderator that the user did not request.
2. Preserve explicit mediation, moderation, serial/parallel mediation, multiple antecedents/outcomes, and controls when the topic clearly requests them. When the requested relation is novel or direct prior evidence is sparse, keep the relation in the study graph and let later hypothesis development use cautious author inference rather than inventing citation support.
3. A moderation relation means source × moderator is associated with target.
4. Keep the graph parsimonious and acyclic for ordinary cross-sectional regression interpretation.
5. Use concise construct labels suitable for a journal article.
6. Put contextual topic terms that are not measured/modelled in background_concepts instead of pretending they are focal variables.
7. The respondent context must be generic and directly relevant. Do NOT invent a country, university, firm, named industry sample frame, software package, date, or institution.
8. This is a cross-sectional observational survey; do not imply causal identification.
9. 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=1200,
                temp=0.0,
                model_role="fast",
                thinking=False,
                allow_fallback=True,
                logical_node='structured_json',
            )
            parsed = agent._extract_json_from_text(raw)
            contract = build_quantitative_contract(topic, keywords, parsed if isinstance(parsed, dict) else fallback_candidate)
            issues = validate_quantitative_contract(contract)
            if issues:
                logging.warning("Adaptive quantitative design validation requires intent-preserving fallback: %s", issues[:3])
                fallback = build_quantitative_contract(topic, keywords, fallback_candidate)
                fallback_issues = validate_quantitative_contract(fallback)
                if fallback_issues:
                    raise RuntimeError(f"quantitative_user_intent_unresolved:{fallback_issues[:3]}")
                return fallback
            context = str(dict(contract.get("context") or {}).get("research_context") or "")
            if re.search(r"\b(university|company|corporation|inc\.?|ltd\.?|country|china|usa|united states|europe|asia)\b", context, re.I):
                fallback_context = "respondents with experience relevant to the focal management phenomenon"
                contract = dict(contract)
                contract["context"] = {"research_context": fallback_context}
                # Rebuild through deterministic normalizer so the hash stays authoritative.
                candidate = parsed if isinstance(parsed, dict) else fallback_candidate
                candidate = dict(candidate)
                candidate["research_context"] = fallback_context
                contract = build_quantitative_contract(topic, keywords, candidate)
            return contract
        except Exception as exc:
            logging.warning("Quantitative adaptive study-design freeze fell back safely: %s", exc)
            return build_quantitative_contract(topic, keywords, fallback_candidate)

    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


    def _compile_graph_evidence(self, mean_scores, contract, bootstrap_iterations=1500):
        """Compile graph-owned associations/indirect effects/moderation deterministically."""
        quant = dict((contract or {}).get("quantitative") or {})
        constructs = {
            str(item.get("construct_id") or ""): str(item.get("label") or "").strip()
            for item in (quant.get("constructs") or []) if isinstance(item, dict)
        }
        relations = [dict(item) for item in (quant.get("relations") or []) if isinstance(item, dict)]
        if not constructs or not relations:
            return {}, pd.DataFrame(), mean_scores.corr(method="spearman")

        # Build one standardized OLS per target so multiple antecedents/controls are estimated together.
        direct_like = [r for r in relations if r.get("type") in {"direct_association", "control_effect"}]
        moderation = [r for r in relations if r.get("type") == "moderation"]
        relation_results = {}
        r2_by_construct = {}
        model_specs = {}
        for target in sorted({r.get("target") for r in relations if r.get("target")}):
            predictors = []
            predictor_meta = []
            for r in direct_like:
                if r.get("target") == target and r.get("source") in constructs:
                    predictors.append(mean_scores[constructs[r["source"]]].to_numpy(dtype=float))
                    predictor_meta.append(("relation", r))
            for r in moderation:
                if r.get("target") != target:
                    continue
                sid, mid = r.get("source"), r.get("moderator")
                if sid not in constructs or mid not in constructs:
                    continue
                x = mean_scores[constructs[sid]].to_numpy(dtype=float)
                m = mean_scores[constructs[mid]].to_numpy(dtype=float)
                # Main effects are included when not already present in the target model.
                existing_sources = {meta[1].get("source") for meta in predictor_meta if meta[0] == "relation"}
                if sid not in existing_sources:
                    predictors.append(x); predictor_meta.append(("main", {"source": sid, "target": target, "label": constructs[sid]}))
                    existing_sources.add(sid)
                if mid not in existing_sources:
                    predictors.append(m); predictor_meta.append(("main", {"source": mid, "target": target, "label": constructs[mid]}))
                    existing_sources.add(mid)
                xz = (x - x.mean()) / (x.std(ddof=0) or 1.0)
                mz = (m - m.mean()) / (m.std(ddof=0) or 1.0)
                predictors.append(xz * mz)
                predictor_meta.append(("interaction", r))
            if not predictors or target not in constructs:
                continue
            y = mean_scores[constructs[target]].to_numpy(dtype=float)
            coef, r2 = self._standardized_ols(y, predictors)
            r2_by_construct[target] = round(float(r2), 3)
            model_specs[target] = (y, predictors, predictor_meta, coef)
            for index, (kind, meta) in enumerate(predictor_meta):
                estimate = round(float(coef[index]), 3)
                if kind == "relation":
                    rid = str(meta.get("relation_id") or "")
                    relation_results[rid] = {
                        "relation_id": rid, "type": meta.get("type"),
                        "source": constructs.get(meta.get("source"), ""),
                        "target": constructs.get(meta.get("target"), ""),
                        "estimate": estimate, "r2_target": round(float(r2), 3),
                    }
                elif kind == "interaction":
                    rid = str(meta.get("relation_id") or "")
                    relation_results[rid] = {
                        "relation_id": rid, "type": "moderation",
                        "source": constructs.get(meta.get("source"), ""),
                        "moderator": constructs.get(meta.get("moderator"), ""),
                        "target": constructs.get(meta.get("target"), ""),
                        "estimate": estimate, "r2_target": round(float(r2), 3),
                        "simple_slope_low": None, "simple_slope_high": None,
                    }

        # Bootstrap relation and indirect-effect CIs from the same compiled model topology.
        rng = np.random.default_rng(20260809)
        relation_boot = {rid: [] for rid in relation_results}
        indirect_boot = {str(e.get("effect_id") or ""): [] for e in (quant.get("indirect_effects") or []) if isinstance(e, dict)}
        n = len(mean_scores)
        for _ in range(int(bootstrap_iterations)):
            idx = rng.integers(0, n, n)
            boot_estimates = {}
            try:
                for target, (y, predictors, predictor_meta, _coef) in model_specs.items():
                    yb = y[idx]
                    xb = [arr[idx] for arr in predictors]
                    coef_b, _ = self._standardized_ols(yb, xb)
                    for j, (kind, meta) in enumerate(predictor_meta):
                        if kind in {"relation", "interaction"}:
                            rid = str(meta.get("relation_id") or "")
                            boot_estimates[rid] = float(coef_b[j])
                            if rid in relation_boot:
                                relation_boot[rid].append(float(coef_b[j]))
                for effect in quant.get("indirect_effects") or []:
                    if not isinstance(effect, dict):
                        continue
                    eid = str(effect.get("effect_id") or "")
                    path = list(effect.get("path") or [])
                    product = 1.0
                    valid = True
                    for a, b in zip(path, path[1:]):
                        rid = next((str(r.get("relation_id") or "") for r in direct_like if r.get("source") == a and r.get("target") == b), "")
                        if not rid or rid not in boot_estimates:
                            valid = False; break
                        product *= boot_estimates[rid]
                    if valid and eid in indirect_boot:
                        indirect_boot[eid].append(float(product))
            except Exception:
                continue

        for rid, values in relation_boot.items():
            if len(values) >= 100:
                lo, hi = np.percentile(values, [2.5, 97.5])
                relation_results[rid]["ci"] = [round(float(lo), 3), round(float(hi), 3)]

        # Recover simple slopes for moderation from standardized main + interaction terms.
        for r in moderation:
            rid = str(r.get("relation_id") or "")
            result = relation_results.get(rid)
            if not result:
                continue
            target = r.get("target")
            _y, _preds, meta_rows, coef = model_specs.get(target, (None, None, [], []))
            main_beta = None
            for j, (kind, meta) in enumerate(meta_rows):
                if kind in {"relation", "main"} and meta.get("source") == r.get("source"):
                    main_beta = float(coef[j]); break
            if main_beta is not None:
                interaction_beta = float(result["estimate"])
                result["simple_slope_low"] = round(main_beta - interaction_beta, 3)
                result["simple_slope_high"] = round(main_beta + interaction_beta, 3)

        indirect_results = {}
        for effect in quant.get("indirect_effects") or []:
            if not isinstance(effect, dict):
                continue
            eid = str(effect.get("effect_id") or "")
            path = list(effect.get("path") or [])
            product = 1.0
            labels = []
            valid = True
            for cid in path:
                labels.append(constructs.get(cid, cid))
            for a, b in zip(path, path[1:]):
                rid = next((str(r.get("relation_id") or "") for r in direct_like if r.get("source") == a and r.get("target") == b), "")
                if not rid or rid not in relation_results:
                    valid = False; break
                product *= float(relation_results[rid]["estimate"])
            if not valid:
                continue
            row = {"effect_id":eid,"path":labels,"estimate":round(product,3),"kind":effect.get("kind") or "simple"}
            vals = indirect_boot.get(eid) or []
            if len(vals) >= 100:
                lo, hi = np.percentile(vals,[2.5,97.5]); row["ci"]=[round(float(lo),3),round(float(hi),3)]
            indirect_results[eid] = row

        path_rows = []
        for r in relations:
            rid = str(r.get("relation_id") or "")
            rr = relation_results.get(rid)
            if not rr:
                continue
            if r.get("type") == "moderation":
                label = f"{rr['source']} × {rr['moderator']} → {rr['target']} (interaction)"
            else:
                label = f"{rr['source']} → {rr['target']}"
            path_rows.append({"Association":label,"Standardized estimate":rr["estimate"],"R² of outcome":rr.get("r2_target","—")})
        for row in indirect_results.values():
            path_rows.append({"Association":"Indirect association: " + " → ".join(row["path"]),"Standardized estimate":row["estimate"],"R² of outcome":"—"})
        path_table = pd.DataFrame(path_rows)

        # Conservative multicollinearity summary across ordinary construct scores.
        vif_values = []
        for target, (_y, predictors, _meta, _coef) in model_specs.items():
            if len(predictors) < 2:
                continue
            matrix = np.column_stack(predictors)
            corr = np.corrcoef(matrix, rowvar=False)
            try:
                inv = np.linalg.pinv(corr)
                vif_values.extend(float(x) for x in np.diag(inv))
            except Exception:
                pass

        summary = {
            "relation_results": relation_results,
            "indirect_results": indirect_results,
            "r2_by_construct": {constructs.get(k,k):v for k,v in r2_by_construct.items()},
            "max_vif": round(max(vif_values),3) if vif_values else 1.0,
            "bootstrap_iterations": int(bootstrap_iterations),
            "study_design_contract_sha256": str((contract or {}).get("contract_sha256") or ""),
        }

        # Exact historical aliases keep legacy validators/checkpoints stable.
        if bool(quant.get("legacy_three_construct_profile")):
            p = next((x for x in quant.get("constructs") or [] if "predictor" in (x.get("roles") or [])), {})
            m = next((x for x in quant.get("constructs") or [] if "mediator" in (x.get("roles") or [])), {})
            o = next((x for x in quant.get("constructs") or [] if "outcome" in (x.get("roles") or [])), {})
            pl, ml, ol = p.get("label"), m.get("label"), o.get("label")
            def est(src,tgt):
                for rr in relation_results.values():
                    if rr.get("source") == src and rr.get("target") == tgt and rr.get("type") == "direct_association": return rr.get("estimate")
                return None
            a, b, direct = est(pl,ml), est(ml,ol), est(pl,ol)
            indirect = next(iter(indirect_results.values()), {})
            summary.update({
                "predictor":pl,"mediator":ml,"outcome":ol,
                "a":a,"b":b,"direct":direct,"indirect":indirect.get("estimate"),
                "r2_mediator":summary["r2_by_construct"].get(ml),
                "r2_outcome":summary["r2_by_construct"].get(ol),
                "indirect_ci":indirect.get("ci"),
                "direct_ci":next((rr.get("ci") for rr in relation_results.values() if rr.get("source")==pl and rr.get("target")==ol), None),
                "predictor_mediator_vif":summary["max_vif"],
            })
        return summary, path_table, mean_scores.corr(method="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 {}
        construct_items = [dict(item) for item in (meta.get("construct_items") or []) if isinstance(item, dict)]
        results = dict(meta.get("association_results") or {}) if isinstance(meta.get("association_results"), dict) else {}
        study_contract = dict(meta.get("study_design_contract") or {}) if isinstance(meta.get("study_design_contract"), dict) else {}
        analysis_plan = list(dict(study_contract.get("quantitative") or {}).get("analysis_plan") or [])
        allowed_methods = [
            "reliability", "composite reliability", "average variance extracted",
            "Fornell-Larcker", "standardized association", "bootstrap",
            "Spearman", "VIF", "descriptive statistics",
        ]
        if "interaction_test" in analysis_plan:
            allowed_methods.extend(["interaction term", "moderation", "simple slopes"])
        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(),
            "study_design_contract": study_contract,
            "constructs": {k: str(v or "").strip() for k, v in constructs.items() if str(v or "").strip()},
            "construct_items": construct_items,
            "association_results": results,
            "uploaded_metric_anchors": uploaded_metric_anchors,
            "causal_identification": "none" if strict else "evidence_owned",
            "allowed_methods": allowed_methods if strict else [],
            "forbidden_unowned_methods": [
                "confirmatory factor analysis", "CFA", "structural equation modeling", "SEM",
                "PLS-SEM", "RMSEA", "CFI", "TLI", "SRMR", "Harman",
                "common method bias", "endogeneity", "instrumental variable",
                "fixed effects", "random effects", "difference-in-differences",
            ] if strict else [],
        }

    @classmethod
    def classify_quantitative_method_threat_claim(cls, sentence, term):
        """Classify a forbidden quantitative term by *claim role*, not token presence.

        R6-R10 separates an actually performed/claimed unowned analysis from a
        limitation, explicit non-use statement, threat discussion, or incidental
        methodological mention.  Only ``PERFORMED_UNOWNED`` is blocking.
        """
        text = re.sub(r"\s+", " ", str(sentence or "")).strip()
        folded = text.casefold()
        label = re.sub(r"\s+", " ", str(term or "")).strip()
        term_folded = label.casefold()
        if not text or not term_folded:
            return {"term": label, "category": "UNKNOWN", "state": "ABSENT"}
        match = re.search(rf"(?i)(?<![A-Za-z]){re.escape(label)}(?![A-Za-z])", text)
        if not match:
            return {"term": label, "category": "UNKNOWN", "state": "ABSENT"}

        threat_terms = {"common method bias", "endogeneity"}
        fit_terms = {"rmsea", "cfi", "tli", "srmr"}
        category = (
            "THREAT" if term_folded in threat_terms else
            "FIT_DIAGNOSTIC" if term_folded in fit_terms else
            "ANALYTIC_METHOD"
        )
        before = folded[max(0, match.start() - 100):match.start()]
        after = folded[match.end():min(len(folded), match.end() + 120)]
        local = folded[max(0, match.start() - 120):min(len(folded), match.end() + 140)]

        scalar_negation = bool(re.search(
            r"\bnot\s+(?:only|merely|just|simply)\b", local
        ))
        explicit_nonuse = (not scalar_negation) and bool(
            re.search(
                r"\b(?:no|not|never|without|did\s+not|does\s+not|do\s+not|was\s+not|were\s+not|is\s+not|are\s+not|cannot|can\s+not)\b[^.;:]{0,55}$",
                before,
            )
            or re.search(
                r"^.{0,45}\b(?:was\s+not|were\s+not|is\s+not|are\s+not|did\s+not|does\s+not|do\s+not|cannot|can\s+not|not\s+(?:used|conducted|performed|estimated|tested|assessed|evaluated|reported|implemented|applied))\b",
                after,
            )
            or re.search(
                r"\b(?:does\s+not|do\s+not|did\s+not|cannot|can\s+not|is\s+not|are\s+not|was\s+not|were\s+not)\s+(?:address|assess|test|establish|identify|rule\s+out|correct|control|substitute|demonstrate|show|prove|support)\b",
                local,
            )
        )
        if explicit_nonuse:
            return {"term": label, "category": category, "state": "NOT_PERFORMED_OR_LIMITATION"}

        performed = bool(
            re.search(
                r"\b(?:we|the\s+study|this\s+study|the\s+analysis|our\s+analysis|the\s+model|the\s+authors?)\b[^.;:]{0,70}\b(?:use|uses|used|employ|employs|employed|conduct|conducts|conducted|perform|performs|performed|estimate|estimates|estimated|test|tests|tested|assess|assesses|assessed|evaluate|evaluates|evaluated|implement|implements|implemented|apply|applies|applied|ran|run|runs|fit|fits|fitted|model|models|modeled|modelled|control|controls|controlled|correct|corrects|corrected|address|addresses|addressed)\b[^.;:]{0,70}$",
                before,
            )
            or re.search(
                r"^.{0,65}\b(?:was|were|is|are)\s+(?:used|employed|conducted|performed|estimated|tested|assessed|evaluated|implemented|applied|fit|fitted|modeled|modelled|controlled|corrected|addressed|reported)\b",
                after,
            )
            or re.search(
                r"^.{0,65}\b(?:estimated|tested|assessed|evaluated|showed|indicated|yielded|produced|confirmed|supported)\b",
                after,
            )
        )
        if category == "FIT_DIAGNOSTIC" and re.search(
            rf"(?i)(?:{re.escape(label)}\s*(?:=|:|was|were)\s*[-+]?\d)|(?:reported|fit\s+index|fit\s+indices)[^.;:]{{0,45}}{re.escape(label)}",
            text,
        ):
            performed = True
        if category == "THREAT":
            # Threats may be discussed as limitations without implying a test.
            if re.search(
                rf"(?i)(?:assess|test|evaluate|examine|address|correct|control)(?:ed|ing|s)?[^.;:]{{0,45}}{re.escape(label)}|{re.escape(label)}[^.;:]{{0,45}}(?:was|were)\s+(?:assessed|tested|evaluated|examined|addressed|corrected|controlled)",
                text,
            ):
                performed = True
            elif not performed:
                return {"term": label, "category": category, "state": "LIMITATION_OR_THREAT"}

        return {
            "term": label,
            "category": category,
            "state": "PERFORMED_UNOWNED" if performed else "MENTION_ONLY",
        }

    @classmethod
    def _first_performed_unowned_quantitative_term(cls, sentence, forbidden_terms):
        for term in forbidden_terms or []:
            frame = cls.classify_quantitative_method_threat_claim(sentence, term)
            if frame.get("state") == "PERFORMED_UNOWNED":
                return frame
        return None

    @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"

            # R6-R12 derived statistical-expression actor frame.  Phrases such
            # as ``the product of the two constituent associations`` and
            # ``multiplying the path coefficients`` denote a deterministic
            # statistical calculation, not a substantive study construct.
            # They receive a distinct actor type so ambiguous reporting verbs
            # (yields/produces/generates/results in) are exempt *only* when the
            # predicate target is also a recognised statistical reporting
            # object.  This keeps construct-level uses such as ``X yields lower
            # anxiety`` fail-closed.
            derived_statistical_actor = (
                r"\b(?:"
                r"(?:the\s+)?product\s+of\s+(?:the\s+)?"
                r"(?:(?:two|constituent|standardized|standardised|estimated|path)\s+){0,4}"
                r"(?:associations?|coefficients?|paths?|estimates?)|"
                r"(?:coefficient|association|path)[- ]product(?:\s+(?:calculation|computation|estimate))?|"
                r"path[- ]product(?:\s+(?:calculation|computation|estimate))?|"
                r"multiplying\s+(?:the\s+)?(?:(?:two|constituent|standardized|standardised|path)\s+){0,4}"
                r"(?:associations?|coefficients?|paths?)|"
                r"multiplication\s+of\s+(?:the\s+)?(?:(?:two|constituent|standardized|standardised|path)\s+){0,4}"
                r"(?:associations?|coefficients?|paths?)"
                r")\b"
            )
            if re.search(derived_statistical_actor, actor, flags=re.I):
                return "DERIVED_STATISTICAL_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"indirect\s+associations?|association\s+estimates?|"
                r"standardized\s+associations?|standardised\s+associations?|"
                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 in {"WEAK_ANALYTICAL_ACTOR", "DERIVED_STATISTICAL_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/R6-R12,
            # such as ``the analysis led to the following numerical estimate``
            # or a derived coefficient product leading to an estimate.
            return actor in {"ANALYTICAL_ACTOR", "WEAK_ANALYTICAL_ACTOR", "DERIVED_STATISTICAL_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"

        # R6-R11 causal-noun scope frame.  A past participle can be either a
        # predicate ("the analysis identified a causal effect") or a modifier
        # inside a causal noun phrase ("rather than an identified causal
        # effect").  The latter inherits the surrounding negation/contrast
        # operator and must not be promoted to an affirmative causal claim.
        causal_noun_participle = (
            r"(?:identified|established|demonstrated|shown|inferred|supported|"
            r"claimed|justified|warranted|proven)"
        )
        causal_noun_adverb = r"(?:empirically|statistically|directly|clearly|credibly|formally|causally)"
        causal_noun_modifier = (
            rf"(?:(?:{causal_noun_adverb}\s+){{0,2}}{causal_noun_participle}\s+)?"
        )
        causal_noun_surface = rf"(?:an?\s+|the\s+)?{causal_noun_modifier}{causal_term}"

        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+{causal_noun_surface}\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)\s+{causal_noun_surface}\b",
            rf"\b(?:is|are|was|were)?\s*(?:restricted|limited|confined)\s+to\b[^.;]*"
            rf"\b(?:not|rather than|instead of)\s+(?:to\s+)?{causal_noun_surface}\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+{causal_noun_surface}\b",
            rf"\b{causal_term}\b\s+(?:is|are|was|were)\s+(?:not|never)\s+"
            rf"{causal_noun_participle}\b",
            rf"\b(?:not|no)\s+{causal_noun_surface}\b",
        )
        limitation_matches = [
            match
            for pattern in limitation_patterns
            for match in re.finditer(pattern, value, flags=re.I)
        ]

        def noun_match_is_inside_limitation(match):
            # Only suppress the affirmative interpretation when the matched
            # predicate/noun phrase is actually inside the frozen limitation
            # span.  A later affirmative clause remains fail-closed.
            return any(
                limited.start() <= match.start() and match.end() <= limited.end()
                for limited in limitation_matches
            )

        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):
                if noun_match_is_inside_limitation(match):
                    continue
                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"

        if saw_disclaimed_direct or limitation_matches:
            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

    @staticmethod
    def _normalize_quantitative_construct_surface(value):
        """Normalize a construct surface for conservative alias matching."""
        text = str(value or "").casefold()
        text = re.sub(r"[^a-z0-9]+", " ", text)
        return re.sub(r"\s+", " ", text).strip()

    @classmethod
    def _quantitative_construct_alias_map(cls, constructs):
        """Return only unique, conservative aliases for frozen construct names.

        R6-R14 permits ordinary abstract compression such as
        ``medical tourism conversion intention`` -> ``conversion intention``.
        Aliases are suffixes of at least two lexical tokens and are admitted only
        when they identify exactly one frozen construct role.  No semantic model
        or fuzzy embedding is used, so ambiguous aliases remain unresolved.
        """
        role_values = {
            str(role): cls._normalize_quantitative_construct_surface(value)
            for role, value in dict(constructs or {}).items()
            if str(value or "").strip()
        }
        candidates = {}
        for role, canonical in role_values.items():
            tokens = canonical.split()
            surfaces = {canonical}
            # Conservative reader-facing shortening: terminal 2-4 token suffixes.
            # One-word aliases (e.g. "intention") are intentionally forbidden.
            for width in range(2, min(4, len(tokens)) + 1):
                if width < len(tokens):
                    alias = " ".join(tokens[-width:])
                    if len(alias) >= 8:
                        surfaces.add(alias)
            for surface in surfaces:
                candidates.setdefault(surface, set()).add(role)
        unique = {surface: next(iter(roles)) for surface, roles in candidates.items() if len(roles) == 1}
        by_role = {role: [] for role in role_values}
        for surface, role in unique.items():
            by_role.setdefault(role, []).append(surface)
        for role in by_role:
            by_role[role] = sorted(set(by_role[role]), key=lambda x: (-len(x), x))
        return by_role

    @classmethod
    def _quantitative_construct_roles_in_text(cls, text, constructs):
        folded = cls._normalize_quantitative_construct_surface(text)
        roles = set()
        aliases = cls._quantitative_construct_alias_map(constructs)
        for role, surfaces in aliases.items():
            for surface in surfaces:
                if re.search(rf"(?<![a-z0-9]){re.escape(surface)}(?![a-z0-9])", folded):
                    roles.add(role)
                    break
        return roles

    @staticmethod
    def _quantitative_metric_local_clause(sentence, metric_start, metric_end):
        """Bound one metric occurrence to its nearest relation clause.

        Strong punctuation/conjunction boundaries are preferred.  If two metric
        occurrences remain in one span, an ``and`` between those metric surfaces
        becomes a boundary.  This avoids treating two path estimates in one
        sentence as if they shared a single owner while not splitting construct
        names such as ``triage and itinerary planning``.
        """
        text = str(sentence or "")
        metric_re = re.compile(r"(?i)(?:β|standardized\s+estimate)\s*(?:=|:)?\s*[-+]?\d+(?:\.\d+)?")
        metrics = list(metric_re.finditer(text))
        boundaries = [0, len(text)]
        for match in re.finditer(r"(?i);|,\s+(?:and|but|while|whereas)\s+|\b(?:however|nevertheless)\b", text):
            boundaries.extend([match.start(), match.end()])
        # Conjunctions between adjacent metric occurrences are relation boundaries.
        for left, right in zip(metrics, metrics[1:]):
            between = text[left.end():right.start()]
            joins = list(re.finditer(r"(?i)\band\b", between))
            if joins:
                join = joins[-1]
                boundaries.extend([left.end() + join.start(), left.end() + join.end()])
        left = max((b for b in boundaries if b <= metric_start), default=0)
        right = min((b for b in boundaries if b >= metric_end), default=len(text))
        clause = text[left:right].strip(" ,;\t\n")
        return clause or text.strip()

    @classmethod
    def _quantitative_metric_owner_for_occurrence(cls, sentence, match, constructs):
        """Resolve a single standardized estimate to a frozen path owner."""
        clause = cls._quantitative_metric_local_clause(sentence, match.start(), match.end())
        roles = cls._quantitative_construct_roles_in_text(clause, constructs)
        folded = cls._normalize_quantitative_construct_surface(clause)
        has_indirect = bool(re.search(r"\bindirect\b", folded))
        has_direct = bool(re.search(r"\b(?:direct|residual direct)\b", folded))

        # Specific semantic labels win only when the construct frame is compatible.
        if has_indirect and {"predictor", "outcome"}.issubset(roles):
            return "indirect", clause
        if has_direct and {"predictor", "outcome"}.issubset(roles):
            return "direct", clause
        if roles == {"predictor", "mediator"}:
            return "a", clause
        if roles == {"mediator", "outcome"}:
            return "b", clause
        if roles == {"predictor", "outcome"} and has_direct:
            return "direct", clause
        # A three-construct clause without an explicit direct/indirect relation is
        # genuinely ambiguous and must not be guessed into a/b ownership.
        return None, clause

    @classmethod
    def _strict_standardized_metric_occurrences(cls, sentence, constructs, results):
        """Yield clause-bound owners for legacy and adaptive graph estimates."""
        regex = re.compile(r"(?i)((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)")
        legacy_expected = {key: results.get(key) for key in ("a", "b", "direct", "indirect")}
        relation_results = dict(results.get("relation_results") or {}) if isinstance(results, dict) else {}
        indirect_results = dict(results.get("indirect_results") or {}) if isinstance(results, dict) else {}
        for match in regex.finditer(str(sentence or "")):
            owner, clause = cls._quantitative_metric_owner_for_occurrence(sentence, match, constructs)
            if owner and legacy_expected.get(owner) is not None:
                yield match, owner, legacy_expected.get(owner), clause
                continue
            folded = cls._normalize_quantitative_construct_surface(clause)
            candidates = []
            for rid, row in relation_results.items():
                if not isinstance(row, dict):
                    continue
                source = cls._normalize_quantitative_construct_surface(row.get("source"))
                target = cls._normalize_quantitative_construct_surface(row.get("target"))
                moderator = cls._normalize_quantitative_construct_surface(row.get("moderator"))
                if not source or not target or source not in folded or target not in folded:
                    continue
                if row.get("type") == "moderation":
                    if not moderator or moderator not in folded or not re.search(r"(?i)interaction|moderat|×", clause):
                        continue
                elif re.search(r"(?i)interaction|moderat|×", clause):
                    continue
                if row.get("estimate") is not None:
                    candidates.append((f"relation:{rid}", row.get("estimate")))
            if re.search(r"(?i)\bindirect\b", clause):
                for eid, row in indirect_results.items():
                    if not isinstance(row, dict):
                        continue
                    labels = [cls._normalize_quantitative_construct_surface(x) for x in (row.get("path") or [])]
                    if labels and all(label in folded for label in labels) and row.get("estimate") is not None:
                        candidates.append((f"indirect:{eid}", row.get("estimate")))
            if len(candidates) == 1:
                yield match, candidates[0][0], candidates[0][1], clause
            else:
                yield match, None, None, clause

    @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)
            method_scope_unknown = (
                strict and ownership == "unknown"
                and bool(re.search(r"(?i)method|measure", str(section_name or "")))
                and cls._first_performed_unowned_quantitative_term(sentence, forbidden) is not None
            )
            if ownership not in {"study", "result"} and not method_scope_unknown:
                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" or method_scope_unknown:
                    method_frame = cls._first_performed_unowned_quantitative_term(sentence, forbidden)
                    if method_frame:
                        violations.append({
                            "type":"unsupported_quantitative_method",
                            "method":method_frame.get("term"),
                            "claim_state":method_frame.get("state"),
                            "method_category":method_frame.get("category"),
                            "preview":sentence.strip()[:180],
                        })

            vif_expected = (results.get("predictor_mediator_vif") if results.get("predictor_mediator_vif") is not None else results.get("max_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)
                for construct_label, expected_r2 in dict(results.get("r2_by_construct") or {}).items():
                    if construct_label and expected_r2 is not None:
                        add_metric_violations(sentence, rf"(R(?:²|\^2)\s*(?:for|of)?\s*{re.escape(str(construct_label))}\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", expected_r2, label=f"r2:{construct_label}")
                # R6-R14 clause-bound metric ownership: validate each occurrence
                # against its own local construct relation, never one owner per sentence.
                for metric_match, metric_owner, metric_expected, metric_clause in cls._strict_standardized_metric_occurrences(
                    sentence, constructs, results
                ):
                    observed = metric_match.group(2)
                    if metric_owner is None or metric_expected is None:
                        violations.append({
                            "type":"quantitative_metric_owner_unresolved",
                            "observed":observed,
                            "preview":str(metric_clause or sentence).strip()[:180],
                            "reason":"The standardized estimate cannot be uniquely bound to a frozen a/b/direct/indirect construct relation.",
                        })
                        continue
                    try:
                        same = abs(float(observed) - float(metric_expected)) < 0.0006
                    except Exception:
                        same = False
                    if not same:
                        expected_text = f"{float(metric_expected):.3f}".rstrip("0").rstrip(".")
                        violations.append({
                            "type":"quantitative_metric_mismatch", "metric":metric_owner,
                            "observed":observed, "expected":expected_text,
                            "preview":str(metric_clause or sentence).strip()[:180],
                        })
            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)
            method_scope_unknown = (
                strict and ownership == "unknown"
                and bool(re.search(r"(?i)method|measure", str(section_name or "")))
                and cls._first_performed_unowned_quantitative_term(sentence, forbidden) is not None
            )
            if ownership not in {"study", "result"} and not method_scope_unknown:
                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" or method_scope_unknown:
                    method_frame = cls._first_performed_unowned_quantitative_term(revised, forbidden)
                    if method_frame:
                        diagnostics.append({
                            "type":"unsupported_quantitative_method_removed",
                            "method":method_frame.get("term"),
                            "claim_state":method_frame.get("state"),
                            "method_category":method_frame.get("category"),
                            "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 results.get("predictor_mediator_vif") is not None else results.get("max_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)
                for construct_label, expected_r2 in dict(results.get("r2_by_construct") or {}).items():
                    if construct_label and expected_r2 is not None:
                        metric_pattern = rf"(R(?:²|\^2)\s*(?:for|of)?\s*{re.escape(str(construct_label))}\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)"
                        replace_local_metric(metric_pattern, expected_r2, label=f"r2:{construct_label}")

                # R6-R14 deterministic repair is also occurrence-local.  Never
                # rewrite every metric in a sentence to one selected path value.
                metric_replacements = []
                for metric_match, metric_owner, metric_expected, metric_clause in cls._strict_standardized_metric_occurrences(
                    revised, constructs, results
                ):
                    if metric_owner is None or metric_expected is None:
                        continue
                    observed = metric_match.group(2)
                    try:
                        same = abs(float(observed) - float(metric_expected)) < 0.0006
                    except Exception:
                        same = False
                    if same:
                        continue
                    expected_text = f"{float(metric_expected):.3f}".rstrip("0").rstrip(".")
                    diagnostics.append({
                        "type":"quantitative_metric_repaired", "metric":metric_owner,
                        "observed":observed, "expected":expected_text,
                        "preview":str(metric_clause or revised).strip()[:180],
                    })
                    metric_replacements.append((metric_match.start(2), metric_match.end(2), expected_text))
                for metric_start, metric_end, replacement in sorted(metric_replacements, reverse=True):
                    revised = revised[:metric_start] + replacement + revised[metric_end:]
            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_contract = self._derive_system_design(agent, topic, keywords)
        issues = validate_quantitative_contract(design_contract)
        if issues:
            raise RuntimeError(f"adaptive_quantitative_study_design_invalid:{issues[:4]}")
        quant = dict(design_contract.get("quantitative") or {})
        construct_items = [dict(item) for item in (quant.get("constructs") or []) if isinstance(item, dict)]
        construct_names = [str(item.get("label") or "").strip() for item in construct_items if str(item.get("label") or "").strip()]
        if len(construct_names) < 2:
            raise RuntimeError("adaptive_quantitative_study_design_has_too_few_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._compile_graph_evidence(mean_scores, design_contract)
        context = str(dict(design_contract.get("context") or {}).get("research_context") or "").strip()

        role_lines = []
        for item in construct_items:
            role_lines.append(
                f"| {item.get('construct_id')} | {item.get('label')} | {', '.join(item.get('roles') or [])} |"
            )
        relation_lines = []
        by_id = {item.get("construct_id"): item.get("label") for item in construct_items}
        for rel in quant.get("relations") or []:
            if not isinstance(rel, dict):
                continue
            source = by_id.get(rel.get("source"), rel.get("source"))
            target = by_id.get(rel.get("target"), rel.get("target"))
            moderator = by_id.get(rel.get("moderator"), rel.get("moderator")) if rel.get("moderator") else "—"
            relation_lines.append(
                f"| {rel.get('relation_id')} | {rel.get('type')} | {source} | {target} | {moderator} |"
            )

        bootstrap_lines = [f"- Bootstrap resamples: {path_summary.get('bootstrap_iterations', 1500)}."]
        for rid, row in dict(path_summary.get("relation_results") or {}).items():
            label = f"{row.get('source')} → {row.get('target')}"
            if row.get("type") == "moderation":
                label = f"{row.get('source')} × {row.get('moderator')} → {row.get('target')}"
            ci = row.get("ci") or []
            ci_text = f"; 95% bootstrap CI [{ci[0]}, {ci[1]}]" if len(ci) == 2 else ""
            extra = ""
            if row.get("type") == "moderation" and row.get("simple_slope_low") is not None:
                extra = f"; standardized simple slopes at moderator -1 SD/+1 SD: {row.get('simple_slope_low')}/{row.get('simple_slope_high')}"
            bootstrap_lines.append(f"- {rid} {label}: standardized estimate {row.get('estimate')}{ci_text}{extra}.")
        for eid, row in dict(path_summary.get("indirect_results") or {}).items():
            ci = row.get("ci") or []
            ci_text = f"; 95% bootstrap CI [{ci[0]}, {ci[1]}]" if len(ci) == 2 else ""
            bootstrap_lines.append(
                f"- {eid} indirect association ({' → '.join(row.get('path') or [])}): {row.get('estimate')}{ci_text}."
            )
        bootstrap_lines.append(f"- Maximum predictor-set VIF across compiled regression models: {path_summary.get('max_vif', 1.0)}.")

        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 |
| Causal identification | None; report associations, indirect associations, and interaction associations only; do not claim causal effects |
| Study design contract | {design_contract.get('contract_id')} |

### Frozen Construct Graph
| Construct ID | Construct | Role(s) |
| :--- | :--- | :--- |
{chr(10).join(role_lines)}

### Frozen Relation Graph
| Relation ID | Type | Source | Target | Moderator |
| :--- | :--- | :--- | :--- | :--- |
{chr(10).join(relation_lines)}

### 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, Interaction, and Collinearity Checks
{chr(10).join(bootstrap_lines)}

[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: The frozen Study Design Contract and relation graph are the sole authority for what this study measures/models. Never add a mediator, moderator, outcome, control, interaction, indirect path, or tested relation that is absent from the graph.
3. INFERENCE DISCIPLINE: This is a cross-sectional observational survey. Use association language. Indirect estimates are not causal mediation identification; interaction estimates are associational moderation tests.
4. MEASUREMENT DEFENSE: State reliability/convergent/discriminant validity only when the displayed values support it.
5. RESULTS DISCIPLINE: Results must report this study's frozen numbers without attaching external citations to coefficients or sample facts.
6. ROBUSTNESS: The Spearman matrix, frozen bootstrap intervals, simple slopes when present, and VIF above are the only frozen robustness evidence. Do not add another robustness procedure.
7. CONSISTENCY: Construct identities, roles, relation IDs, sample size, context, scale, and all numerical results above are the single source of quantitative truth throughout the manuscript.
"""

        role_map = {}
        for role in ("predictor", "mediator", "moderator", "outcome", "control", "covariate"):
            vals = [item.get("label") for item in construct_items if role in (item.get("roles") or [])]
            if len(vals) == 1:
                role_map[role] = vals[0]

        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,
            "study_design_contract": design_contract,
            "constructs": role_map,
            "construct_items": construct_items,
            "association_results": path_summary,
            "method": (
                "Positive-semidefinite latent correlation generation, indicator loadings, Likert discretization, "
                "reliability/validity summaries, Study-Design-Contract-compiled standardized association models, "
                "fixed-seed bootstrap intervals, interaction/simple-slope estimates when frozen, and Spearman robustness correlations."
            ),
            "generation_parameters": {
                "constructs": construct_names,
                "study_design_contract_sha256": design_contract.get("contract_sha256"),
                "analysis_plan": list(quant.get("analysis_plan") or []),
                "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.get("bootstrap_iterations", 1500),
                "bootstrap_seed": 20260809,
                "reported_summaries": [
                    "Cronbach alpha", "composite reliability", "average variance extracted",
                    "descriptive statistics", "Fornell-Larcker matrix", "compiled standardized association estimates",
                    "R-squared", "bootstrap confidence intervals", "interaction/simple-slope estimates when frozen",
                    "Spearman correlations", "VIF",
                ],
            },
            "evidence_sections": [
                "Frozen quantitative study design", "Frozen construct graph", "Frozen relation graph",
                "Reliability and convergent validity", "Descriptive statistics and discriminant validity",
                "Frozen association model results", "Rank-order robustness correlations",
                "Frozen bootstrap, interaction, and collinearity checks",
            ],
        }
        metadata["semantic_claim_frame_version"] = self.SEMANTIC_CLAIM_FRAME_VERSION
        metadata["front_matter_contract"] = self.build_front_matter_contract(topic, keywords, design_contract)
        metadata["hypothesis_ledger"] = self.build_hypothesis_ledger(design_contract)
        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()