A5-R3-R7-R2C-R2 adaptive study design scoped deployment

core/generation_runner.py

@@ -998,15 +998,23 @@ class GenerationRunner:
             diagnostics = quant_engine.inspect_quantitative_study_fact_contract(
                 value, data_metadata, section_name=section_name
             )
-        return value, list(diagnostics or [])
+        diagnostics = list(diagnostics or [])
+        scope_inspector = getattr(agent, "inspect_study_design_scope_contract", None)
+        if callable(scope_inspector):
+            diagnostics.extend(scope_inspector(value, data_metadata, section_name=section_name) or [])
+        return value, diagnostics
 
     @staticmethod
     def _study_fact_violations(mode: str, agent, quant_engine, text: str, data_metadata: dict, *, section_name: str):
+        rows = []
         if str(mode or '') == 'asq':
-            return list(agent.inspect_qualitative_study_fact_contract(text, data_metadata, section_name=section_name) or [])
-        if str(mode or '') == 'amj':
-            return list(quant_engine.inspect_quantitative_study_fact_contract(text, data_metadata, section_name=section_name) or [])
-        return []
+            rows.extend(agent.inspect_qualitative_study_fact_contract(text, data_metadata, section_name=section_name) or [])
+        elif str(mode or '') == 'amj':
+            rows.extend(quant_engine.inspect_quantitative_study_fact_contract(text, data_metadata, section_name=section_name) or [])
+        scope_inspector = getattr(agent, "inspect_study_design_scope_contract", None)
+        if callable(scope_inspector):
+            rows.extend(scope_inspector(text, data_metadata, section_name=section_name) or [])
+        return list(rows)
 
     @staticmethod
     def _section_retry_feedback(violations, *, limit=8) -> str:
@@ -1072,6 +1080,10 @@ class GenerationRunner:
             "hypothesis_contract_missing_id", "hypothesis_contract_ordinal_alias",
             "hypothesis_contract_pairing_mismatch", "hypothesis_contract_unknown_id",
             "citation_outside_frozen_section_whitelist", "authorial_ai_meta_trace",
+            "study_design_scope_mismatch", "unsupported_focal_construct", "unsupported_relation_claim",
+            "unsupported_moderation_claim", "unsupported_mediation_claim", "unsupported_temporal_scope",
+            "unsupported_design_tradition", "unsupported_unit_of_analysis", "unsupported_data_source_claim",
+            "unsupported_process_claim",
         }
         if kind in always_fatal:
             return "FATAL"
@@ -1115,6 +1127,12 @@ class GenerationRunner:
             "hypothesis": {"hypothesis_contract_missing_id", "hypothesis_contract_ordinal_alias",
                            "hypothesis_contract_pairing_mismatch", "hypothesis_contract_unknown_id"},
             "workflow_trace": {"authorial_ai_meta_trace", "author_facing_internal_workflow_language"},
+            "study_design_scope": {
+                "study_design_scope_mismatch", "unsupported_focal_construct", "unsupported_relation_claim",
+                "unsupported_moderation_claim", "unsupported_mediation_claim", "unsupported_temporal_scope",
+                "unsupported_design_tradition", "unsupported_unit_of_analysis", "unsupported_data_source_claim",
+                "unsupported_process_claim",
+            },
         }
         for lineage, kinds in groups.items():
             if kind in kinds:
@@ -1438,8 +1456,13 @@ class GenerationRunner:
             hid = str(item.get("hypothesis_id") or "").strip()
             predictor = str(item.get("predictor") or "").strip()
             outcome = str(item.get("outcome") or "").strip()
+            moderator = str(item.get("moderator") or "").strip()
+            relation = str(item.get("relation") or "association").strip().casefold()
             if hid and predictor and outcome:
-                lines.append(f"{hid}: {predictor} is associated with {outcome}.")
+                if relation == "moderation" and moderator:
+                    lines.append(f"{hid}: {moderator} moderates the association between {predictor} and {outcome}.")
+                else:
+                    lines.append(f"{hid}: {predictor} is associated with {outcome}.")
         if not lines:
             return ""
         indirect = dict(record.get("indirect_analysis") or {})
@@ -1470,8 +1493,13 @@ class GenerationRunner:
             hid = str(item.get("hypothesis_id") or "").strip().upper()
             predictor = re.sub(r"\s+", " ", str(item.get("predictor") or "")).strip()
             outcome = re.sub(r"\s+", " ", str(item.get("outcome") or "")).strip()
+            moderator = str(item.get("moderator") or "").strip()
+            relation = str(item.get("relation") or "association").strip().casefold()
             if hid and predictor and outcome:
-                lines.append(f"{hid}: {predictor} is associated with {outcome}.")
+                if relation == "moderation" and moderator:
+                    lines.append(f"{hid}: {moderator} moderates the association between {predictor} and {outcome}.")
+                else:
+                    lines.append(f"{hid}: {predictor} is associated with {outcome}.")
         return "\n".join(lines)
 
     @staticmethod
@@ -1526,7 +1554,7 @@ class GenerationRunner:
             })
         all_constructs = {
             re.sub(r"\s+", " ", str(item.get(key) or "")).strip()
-            for item in items for key in ("predictor", "outcome")
+            for item in items for key in ("predictor", "outcome", "moderator")
             if str(item.get(key) or "").strip()
         }
         for start, end, sentence in SSciPipelineAgent._sentence_spans(value):
@@ -1543,6 +1571,7 @@ class GenerationRunner:
                 continue
             predictor = re.sub(r"\s+", " ", str(item.get("predictor") or "")).strip()
             outcome = re.sub(r"\s+", " ", str(item.get("outcome") or "")).strip()
+            moderator = re.sub(r"\s+", " ", str(item.get("moderator") or "")).strip()
             lower_sentence = re.sub(r"\s+", " ", sentence).casefold()
             mentioned_constructs = {
                 term for term in all_constructs if term.casefold() in lower_sentence
@@ -1550,8 +1579,9 @@ class GenerationRunner:
             # Pairing identity is global, not limited to the theory-definition
             # sections. Require at least two named constructs so a bare reference
             # such as "H1 was supported" cannot be misclassified as remapping.
-            if len(mentioned_constructs) >= 2 and predictor and outcome and (
-                predictor.casefold() not in lower_sentence or outcome.casefold() not in lower_sentence
+            expected_terms = [term for term in (predictor, outcome, moderator) if term]
+            if len(mentioned_constructs) >= 2 and predictor and outcome and any(
+                term.casefold() not in lower_sentence for term in expected_terms
             ):
                 rows.append({
                     "type": "hypothesis_contract_pairing_mismatch",
@@ -3189,10 +3219,14 @@ class GenerationRunner:
                 hypothesis_ledger = dict(data_metadata.get("hypothesis_ledger") or {})
                 expected_hypothesis_version = str(getattr(quant_engine, "HYPOTHESIS_LEDGER_VERSION", ""))
                 if hypothesis_ledger.get("version") != expected_hypothesis_version:
-                    constructs = dict(data_metadata.get("constructs") or {})
-                    rebuilt_hypotheses = quant_engine.build_hypothesis_ledger([
-                        constructs.get("predictor"), constructs.get("mediator"), constructs.get("outcome")
-                    ])
+                    study_contract = dict(data_metadata.get("study_design_contract") or {})
+                    if study_contract:
+                        rebuilt_hypotheses = quant_engine.build_hypothesis_ledger(study_contract)
+                    else:
+                        constructs = dict(data_metadata.get("constructs") or {})
+                        rebuilt_hypotheses = quant_engine.build_hypothesis_ledger([
+                            constructs.get("predictor"), constructs.get("mediator"), constructs.get("outcome")
+                        ])
                     if not rebuilt_hypotheses:
                         raise RuntimeError("quantitative_hypothesis_ledger_rebuild_failed")
                     data_metadata["hypothesis_ledger"] = rebuilt_hypotheses
@@ -3203,12 +3237,16 @@ class GenerationRunner:
                 front_contract = dict(data_metadata.get("front_matter_contract") or {})
                 expected_front_version = str(getattr(quant_engine, "FRONT_MATTER_CONTRACT_VERSION", ""))
                 if front_contract.get("version") != expected_front_version:
-                    constructs = dict(data_metadata.get("constructs") or {})
-                    rebuilt_front = quant_engine.build_front_matter_contract(
-                        topic, keywords, [
-                            constructs.get("predictor"), constructs.get("mediator"), constructs.get("outcome")
-                        ]
-                    )
+                    study_contract = dict(data_metadata.get("study_design_contract") or {})
+                    if study_contract:
+                        rebuilt_front = quant_engine.build_front_matter_contract(topic, keywords, study_contract)
+                    else:
+                        constructs = dict(data_metadata.get("constructs") or {})
+                        rebuilt_front = quant_engine.build_front_matter_contract(
+                            topic, keywords, [
+                                constructs.get("predictor"), constructs.get("mediator"), constructs.get("outcome")
+                            ]
+                        )
                     if rebuilt_front:
                         data_metadata["front_matter_contract"] = rebuilt_front
                     else:
@@ -4189,7 +4227,7 @@ class GenerationRunner:
                 table_rule = (
                     "AUTHORITATIVE TABLE CONTRACT: Do not create, reproduce, renumber, or alter any Markdown statistical table. "
                     "Required statistical tables are handled separately from narrative prose. "
-                    "You may discuss and refer to Table 1-Table 4 where relevant, but do not invent an additional table or change a table number."
+                    f"You may discuss and refer only to the frozen table numbers {[int(row.get('number') or 0) for row in data_metadata.get('quantitative_table_manifest', {}).get('tables', []) if int(row.get('number') or 0) > 0]}; do not invent an additional table or change a table number."
                 )
             elif sec["is_method_or_result"] and mode == "amj":
                 table_rule = "Present only study-supported statistical evidence; do not invent tables or numerical results."

core/pipeline_agent.py

@@ -28,6 +28,7 @@ except ImportError:
 
 from config import APP_CONFIG
 from core.production_guard import find_authorial_ai_meta_spans
+from core.study_design_contract import inspect_study_design_scope
 from core.model_gateway import ProviderNeutralModelGateway
 from core.model_usage import ModelUsageEvent, emit_usage_event, normalize_deepseek_usage, prompt_identity, sanitize_metadata
 
@@ -1761,6 +1762,30 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
                     )
         raise DeepSeekRequestError(f'{phase} front matter validation failed: {last_error}')
 
+    @staticmethod
+    def _quantitative_front_matter_items(contract):
+        """Return extensible frozen construct rows, with legacy role-map fallback."""
+        record = dict(contract or {})
+        items = [dict(item) for item in (record.get("construct_items") or []) if isinstance(item, dict)]
+        if items:
+            return [
+                {"role": "/".join(item.get("roles") or []), "name": re.sub(r"\s+", " ", str(item.get("label") or "")).strip()}
+                for item in items if str(item.get("label") or "").strip()
+            ]
+        constructs = dict(record.get("constructs") or {})
+        roles = list(record.get("construct_roles") or constructs.keys())
+        return [
+            {"role": role, "name": re.sub(r"\s+", " ", str(constructs.get(role) or "")).strip()}
+            for role in roles if str(constructs.get(role) or "").strip()
+        ]
+
+    @staticmethod
+    def inspect_study_design_scope_contract(text, metadata, section_name=""):
+        contract = dict((metadata or {}).get("study_design_contract") or {})
+        if not contract:
+            return []
+        return inspect_study_design_scope(text, contract, section_name=section_name)
+
     @staticmethod
     def _quantitative_front_matter_fallback(contract):
         """Last-resort degraded front matter derived only from frozen authority.
@@ -1769,10 +1794,8 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
         semantic reconciliation cannot produce a validated result; it never
         replaces the normal LLM path.
         """
-        constructs = dict((contract or {}).get("constructs") or {})
-        roles = list((contract or {}).get("construct_roles") or ("predictor", "mediator", "outcome"))
-        values = [re.sub(r"\s+", " ", str(constructs.get(role) or "")).strip() for role in roles]
-        values = [item for item in values if item]
+        items = SSciPipelineAgent._quantitative_front_matter_items(contract)
+        values = [row.get("name") for row in items if row.get("name")]
         if not values:
             return {"title": "Quantitative Management Study", "keywords": ["quantitative research", "management research", "empirical study", "evidence"]}
         title = ", ".join(values[:-1]) + ((", and " if len(values) > 2 else " and ") + values[-1] if len(values) > 1 else "")
@@ -1802,14 +1825,13 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
             issues.append("keywords_count_must_be_4_to_6")
         if len({item.casefold() for item in keywords}) != len(keywords):
             issues.append("keywords_must_be_distinct")
-        constructs = dict((contract or {}).get("constructs") or {})
-        roles = list((contract or {}).get("construct_roles") or constructs.keys())
+        frozen_items = SSciPipelineAgent._quantitative_front_matter_items(contract)
         combined = (title + " ; " + " ; ".join(keywords)).casefold()
         missing = [
-            str(constructs.get(role) or "").strip()
-            for role in roles
-            if str(constructs.get(role) or "").strip()
-            and str(constructs.get(role) or "").strip().casefold() not in combined
+            str(row.get("name") or "").strip()
+            for row in frozen_items
+            if str(row.get("name") or "").strip()
+            and str(row.get("name") or "").strip().casefold() not in combined
         ]
         if missing:
             issues.append("missing_frozen_constructs:" + " | ".join(missing))
@@ -1825,9 +1847,7 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
         degraded-mode fallback keeps packaging recoverable.
         """
         contract = dict(contract or {})
-        constructs = dict(contract.get("constructs") or {})
-        roles = list(contract.get("construct_roles") or constructs.keys())
-        frozen = [{"role": role, "name": constructs.get(role)} for role in roles if constructs.get(role)]
+        frozen = self._quantitative_front_matter_items(contract)
         if not frozen:
             logging.warning("Front-matter semantic contract missing frozen constructs; using degraded fallback.")
             return self._quantitative_front_matter_fallback(contract)

core/qualitative_engine.py

@@ -9,12 +9,14 @@ import logging
 import re
 
 from config import APP_CONFIG
+from core.study_design_contract import STUDY_DESIGN_CONTRACT_VERSION, build_qualitative_contract
 
 
 class QualitativeEngine:
     QUALITATIVE_APPENDIX_MANIFEST_VERSION = "qualitative-appendix-manifest-v1"
     TARGET_APPENDIX_BINDING_VERSION = "target-qualitative-appendix-binding-v2"
     SEMANTIC_CLAIM_FRAME_VERSION = "smartpaper-core-semantic-claim-frame-v1"
+    STUDY_DESIGN_CONTRACT_VERSION = STUDY_DESIGN_CONTRACT_VERSION
     APPENDIX_FIELDS = ("theme", "concept", "stage", "case_label", "evidence_role")
     APPENDIX_CANONICAL_LABEL_FIELDS = ("theme", "concept")
     APPENDIX_HEADERS = (
@@ -982,6 +984,7 @@ Rules:
             "evidence": clean_evidence,
             "themes": themes,
         }
+        metadata["study_design_contract"] = build_qualitative_contract(topic, keywords, metadata)
         metadata["semantic_claim_frame"] = self.build_qualitative_semantic_claim_frame(metadata)
         status_message = "✅ 定性研究事实、分析设计与证据结构已冻结。"
         return md, status_message, metadata

core/quantitative_engine.py

@@ -13,6 +13,13 @@ 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,
+)
 
 
 class QuantitativeEngine:
@@ -24,6 +31,7 @@ class QuantitativeEngine:
     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": {
@@ -111,70 +119,34 @@ class QuantitativeEngine:
 
     @classmethod
     def build_front_matter_contract(cls, topic, keywords, constructs):
-        """Freeze semantic authority without hard-coding reader-facing prose.
+        """Build a reader-front-matter semantic contract from adaptive design authority.
 
-        The evidence engine owns construct identities/roles.  The user's topic
-        and keyword seeds remain available for retrieval, literature synthesis,
-        and domain reasoning.  Reader-facing title/keywords are reconciled later
-        by the model against this frozen contract and then deterministically
-        validated.
+        ``constructs`` accepts either the new Study Design Contract or the legacy
+        three-construct list/dict so old checkpoints remain readable.
         """
-        role_names = ("predictor", "mediator", "outcome")
+        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):
-            role_map = {
-                role: re.sub(r"\s+", " ", str(constructs.get(role) or "")).strip()
-                for role in role_names
-            }
+            values = [constructs.get(role) for role in ("predictor", "mediator", "outcome")]
         else:
-            values = [re.sub(r"\s+", " ", str(item or "")).strip() for item in (constructs or [])]
-            if len(values) != len(role_names):
-                return {}
-            role_map = dict(zip(role_names, values))
-        if any(not role_map.get(role) for role in role_names):
-            return {}
-        source_keywords = [
-            re.sub(r"\s+", " ", item).strip()
-            for item in re.split(r"[,;\n|]+", str(keywords or ""))
-            if re.sub(r"\s+", " ", item).strip()
-        ]
-        return {
-            "version": cls.FRONT_MATTER_CONTRACT_VERSION,
-            "policy": "system_generated_frozen_construct_semantic_authority",
-            "source_topic": re.sub(r"\s+", " ", str(topic or "")).strip(),
-            "source_keywords": source_keywords,
-            "construct_roles": list(role_names),
-            "constructs": role_map,
-        }
+            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):
-        """Freeze one result-independent AMJ hypothesis identity contract.
-
-        The system-generated quantitative design is observational and does not
-        justify result-driven directional hypotheses.  The three numbered
-        hypotheses therefore freeze only the construct pairing/association
-        identity.  The indirect association is an analysis target, not H4.
-        """
-        values = [re.sub(r"\s+", " ", str(item or "")).strip() for item in (constructs or [])]
-        if len(values) != 3 or any(not item for item in values):
-            return {}
-        predictor, mediator, outcome = values
-        items = [
-            {"hypothesis_id": "H1", "predictor": predictor, "outcome": outcome, "relation": "association"},
-            {"hypothesis_id": "H2", "predictor": predictor, "outcome": mediator, "relation": "association"},
-            {"hypothesis_id": "H3", "predictor": mediator, "outcome": outcome, "relation": "association"},
-        ]
-        return {
-            "version": cls.HYPOTHESIS_LEDGER_VERSION,
-            "items": items,
-            "indirect_analysis": {
-                "predictor": predictor,
-                "mediator": mediator,
-                "outcome": outcome,
-                "numbered_hypothesis": False,
-                "relation": "estimated_indirect_association",
-            },
-        }
+        """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):
@@ -478,42 +450,72 @@ class QuantitativeEngine:
 
     @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[:3]
+        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=""):
-        """Freeze a parsimonious topic-aligned design without inventing named sites."""
-        fallback = {
+        """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.
+        """
+        fallback_candidate = {
             "constructs": [
-                "Focal Predictor",
-                "Focal Mediator",
-                "Focal Outcome",
+                {"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},
             ],
-            "research_context": "respondents with experience relevant to the focal management phenomenon",
+            "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 fallback
-        prompt = f"""Design the minimum coherent quantitative management-study skeleton for the topic below.
-Return ONLY valid JSON:
-{{"constructs":["predictor construct","mediating construct","outcome construct"],"research_context":"one generic respondent context"}}
+            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:
-1. Exactly three distinct constructs in predictor -> mediator -> outcome order.
-2. Use concise construct labels that can appear throughout a journal article.
-3. The context must be generic and directly relevant to the topic. Do NOT name or invent a country, university, firm, industry sample frame, software package, time period, or institution.
-4. This is a cross-sectional single-wave survey design, so do not imply experimental or causal identification.
-5. Do not create any number; sample size and scale are frozen elsewhere.
+1. Use 2-8 distinct constructs only when the topic actually requires them; do NOT force a mediator or moderator.
+2. Preserve explicit mediation, moderation, serial/parallel mediation, multiple antecedents/outcomes, and controls when the topic clearly requests them.
+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=500,
+                max_tokens=1200,
                 temp=0.0,
                 model_role="fast",
                 thinking=False,
@@ -521,23 +523,25 @@ Keywords: {keywords}"""
                 logical_node='structured_json',
             )
             parsed = agent._extract_json_from_text(raw)
-            if not isinstance(parsed, dict):
-                return fallback
-            constructs = self._safe_constructs(parsed.get("constructs"))
-            context = re.sub(r"\s+", " ", str(parsed.get("research_context") or "")).strip()
-            if len(constructs) != 3:
-                return fallback
-            if not context or len(context) > 240:
-                context = fallback["research_context"]
-            # Named-location/site details are deliberately not accepted from this
-            # design step. The manuscript may remain generic when no user evidence
-            # supplies a real setting.
+            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 fell back safely: %s", issues[:3])
+                return build_quantitative_contract(topic, keywords, fallback_candidate)
+            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):
-                context = fallback["research_context"]
-            return {"constructs": constructs, "research_context": context}
+                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 study-design freeze fell back safely: %s", exc)
-            return fallback
+            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,
@@ -712,6 +716,220 @@ Keywords: {keywords}"""
         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."""
@@ -875,7 +1093,17 @@ Keywords: {keywords}"""
             }
 
         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,
@@ -883,15 +1111,13 @@ Keywords: {keywords}"""
             "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": [
-                "reliability", "composite reliability", "average variance extracted",
-                "Fornell-Larcker", "standardized association", "bootstrap",
-                "Spearman", "VIF", "descriptive statistics",
-            ] if strict else [],
+            "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",
@@ -1601,12 +1827,44 @@ Keywords: {keywords}"""
 
     @classmethod
     def _strict_standardized_metric_occurrences(cls, sentence, constructs, results):
-        """Yield clause-bound owners for each beta/standardized-estimate occurrence."""
+        """Yield clause-bound owners for legacy and adaptive graph estimates."""
         regex = re.compile(r"(?i)((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)")
-        expected = {key: results.get(key) for key in ("a", "b", "direct", "indirect")}
+        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)
-            yield match, owner, expected.get(owner) if owner else None, clause
+            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)indirect", 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=""):
@@ -1715,7 +1973,7 @@ Keywords: {keywords}"""
                             "preview":sentence.strip()[:180],
                         })
 
-            vif_expected = results.get("predictor_mediator_vif") if strict else uploaded.get("VIF")
+            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")
@@ -1724,6 +1982,9 @@ Keywords: {keywords}"""
                     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(
@@ -1891,7 +2152,7 @@ Keywords: {keywords}"""
                     return match.group(1) + expected_text
                 revised = regex.sub(repl, revised)
 
-            vif_expected = results.get("predictor_mediator_vif") if strict else uploaded.get("VIF")
+            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")
@@ -1902,6 +2163,10 @@ Keywords: {keywords}"""
                     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.
@@ -1969,18 +2234,14 @@ Keywords: {keywords}"""
             if user_md:
                 evidence_sections = [
                     heading.strip()
-                    for heading in re.findall(
-                        r"(?m)^#{1,3}\s+(.+?)\s*$",
-                        user_md,
-                    )
+                    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_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."
@@ -1989,13 +2250,58 @@ Keywords: {keywords}"""
                 metadata["quantitative_evidence_ledger"] = self.build_quantitative_evidence_ledger(metadata, user_md)
                 return None, user_md, tr(language, "actual_stats_parsed"), metadata
 
-        design = self._derive_system_design(agent, topic, keywords)
-        construct_names = design["constructs"]
-        df, validity_df, fl_matrix_df, mean_scores, corr_matrix = self.generate_empirical_data_robust(
+        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._path_evidence(mean_scores, construct_names)
-        context = design["research_context"]
+        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 |
 | :--- | :--- |
@@ -2003,10 +2309,18 @@ Keywords: {keywords}"""
 | Respondent context | {context} |
 | Sample size | {len(df)} |
 | Response scale | 7-point Likert-type scale |
-| Predictor | {construct_names[0]} |
-| Mediator | {construct_names[1]} |
-| Outcome | {construct_names[2]} |
-| Causal identification | None; report associations and estimated indirect association, not causal effects |
+| 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)}
@@ -2021,20 +2335,25 @@ Keywords: {keywords}"""
 ### Table 4: Rank-Order Robustness Correlations (Spearman)
 {spearman.round(3).to_markdown()}
 
-### Frozen Bootstrap and Collinearity Checks
-- Bootstrap resamples: {path_summary['bootstrap_iterations']}.
-- Indirect association ({construct_names[0]} → {construct_names[1]} → {construct_names[2]}): {path_summary['indirect']}; 95% bootstrap CI [{path_summary['indirect_ci'][0]}, {path_summary['indirect_ci'][1]}].
-- Residual direct association ({construct_names[0]} → {construct_names[2]} controlling {construct_names[1]}): {path_summary['direct']}; 95% bootstrap CI [{path_summary['direct_ci'][0]}, {path_summary['direct_ci'][1]}].
-- Predictor/mediator VIF in the outcome model: {path_summary['predictor_mediator_vif']}.
+### 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: This is a cross-sectional observational survey. Use association language. The indirect estimate is not causal mediation identification.
-3. MEASUREMENT DEFENSE: State reliability/convergent/discriminant validity only when the displayed values support it.
-4. RESULTS DISCIPLINE: Results must report this study's frozen numbers without attaching external citations to the coefficients or sample facts.
-5. ROBUSTNESS: The Spearman matrix, bootstrap intervals, and VIF above are the only frozen robustness evidence. Do not add another robustness procedure.
-6. CONSISTENCY: The predictor, mediator, outcome, sample size, context, scale, and all numerical results above are the single source of quantitative truth throughout the manuscript.
+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",
@@ -2043,19 +2362,19 @@ Keywords: {keywords}"""
             "scale_points": 7,
             "study_design": "cross-sectional single-wave questionnaire",
             "research_context": context,
-            "constructs": {
-                "predictor": construct_names[0],
-                "mediator": construct_names[1],
-                "outcome": construct_names[2],
-            },
+            "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, standardized association "
-                "models, fixed-seed bootstrap intervals, and Spearman robustness correlations."
+                "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",
@@ -2063,45 +2382,28 @@ Keywords: {keywords}"""
                 "latent_correlation_control": "positive-semidefinite matrix with maximum target correlation 0.55",
                 "item_generation": "loading × latent score + normally distributed measurement error",
                 "discretization": "round to nearest integer and clamp to 1–7",
-                "bootstrap_iterations": path_summary["bootstrap_iterations"],
+                "bootstrap_iterations": path_summary.get("bootstrap_iterations", 1500),
                 "bootstrap_seed": 20260809,
                 "reported_summaries": [
-                    "Cronbach alpha",
-                    "composite reliability",
-                    "average variance extracted",
-                    "descriptive statistics",
-                    "Fornell-Larcker matrix",
-                    "standardized association estimates",
-                    "R-squared",
-                    "bootstrap confidence intervals",
-                    "Spearman correlations",
-                    "VIF",
+                    "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",
-                "Reliability and convergent validity",
-                "Descriptive statistics and discriminant validity",
-                "Frozen association model results",
-                "Rank-order robustness correlations",
-                "Frozen bootstrap and collinearity checks",
+                "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, metadata.get("constructs") or construct_names
-        )
-        metadata["hypothesis_ledger"] = self.build_hypothesis_ledger(construct_names)
-        metadata["quantitative_table_manifest"] = self.build_table_manifest(
-            validity_df, fl_matrix_df, path_table, spearman
-        )
+        metadata["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,
-        )
+        return df, generated_prompt, tr(language, "synthetic_stats_generated"), metadata
 
 
 def get_quantitative_engine():

core/study_design_contract.py

@@ -0,0 +1,496 @@
+# -*- coding: utf-8 -*-
+"""Adaptive study-design contracts shared by SmartPaper quantitative/qualitative paths.
+
+The LLM may propose a candidate design, but this module owns deterministic
+normalization, identity, validation, and scope semantics.  It deliberately
+contains no provider/model routing logic.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from typing import Any
+
+STUDY_DESIGN_CONTRACT_VERSION = "smartpaper-adaptive-study-design-contract-v1"
+
+_QUANT_ROLES = {"predictor", "mediator", "moderator", "outcome", "control", "covariate"}
+_QUANT_RELATIONS = {"direct_association", "moderation", "control_effect"}
+_QUAL_SOURCE_FAMILIES = {
+    "interview", "focus_group", "observation", "document", "log",
+    "artifact", "digital_trace", "field_note", "other",
+}
+
+
+def _clean(value: Any, maximum: int = 240) -> str:
+    text = re.sub(r"\s+", " ", str(value or "")).strip(" \t\r\n,;:.-")
+    return text[:maximum]
+
+
+def _keywords(value: Any) -> list[str]:
+    raw = value if isinstance(value, (list, tuple)) else re.split(r"[,;\n|]+", str(value or ""))
+    out: list[str] = []
+    seen: set[str] = set()
+    for item in raw:
+        text = _clean(item, 100)
+        key = text.casefold()
+        if text and key not in seen:
+            out.append(text)
+            seen.add(key)
+    return out[:16]
+
+
+def _stable_hash(payload: dict[str, Any]) -> str:
+    encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _finalize(contract: dict[str, Any]) -> dict[str, Any]:
+    value = dict(contract)
+    value.pop("contract_sha256", None)
+    value.pop("contract_id", None)
+    digest = _stable_hash(value)
+    value["contract_id"] = digest[:20]
+    value["contract_sha256"] = digest
+    return value
+
+
+def _normalize_quant_relation_type(value: Any) -> str:
+    text = _clean(value, 80).casefold().replace("-", "_").replace(" ", "_")
+    aliases = {
+        "association": "direct_association", "direct": "direct_association",
+        "direct_effect": "direct_association", "direct_association": "direct_association",
+        "path": "direct_association", "mediation_path": "direct_association",
+        "moderator": "moderation", "moderation": "moderation", "interaction": "moderation",
+        "interaction_effect": "moderation", "control": "control_effect", "control_effect": "control_effect",
+    }
+    return aliases.get(text, text)
+
+
+def _normalize_roles(value: Any) -> list[str]:
+    raw = value if isinstance(value, (list, tuple)) else [value]
+    out: list[str] = []
+    for item in raw:
+        role = _clean(item, 40).casefold().replace(" ", "_")
+        if role in _QUANT_ROLES and role not in out:
+            out.append(role)
+    return out
+
+
+def _legacy_quant_candidate(constructs: list[str], context: str = "") -> dict[str, Any]:
+    values = [_clean(item, 80) for item in constructs if _clean(item, 80)]
+    if len(values) != 3:
+        values = ["Focal Predictor", "Focal Mediator", "Focal Outcome"]
+    p, m, o = values
+    return {
+        "constructs": [
+            {"label": p, "roles": ["predictor"]},
+            {"label": m, "roles": ["mediator"]},
+            {"label": o, "roles": ["outcome"]},
+        ],
+        "relations": [
+            {"source": p, "target": o, "type": "direct_association", "hypothesis_order": 1},
+            {"source": p, "target": m, "type": "direct_association", "hypothesis_order": 2},
+            {"source": m, "target": o, "type": "direct_association", "hypothesis_order": 3},
+        ],
+        "indirect_effects": [{"path": [p, m, o], "kind": "simple", "numbered_hypothesis": False}],
+        "research_context": context,
+        "background_concepts": [],
+    }
+
+
+def build_quantitative_contract(
+    topic: str,
+    keywords: Any,
+    candidate: Any,
+    *,
+    source_type: str = "system_generated",
+    policy_version: str = "adaptive-quantitative-v1",
+) -> dict[str, Any]:
+    """Normalize a model-proposed quantitative graph into deterministic authority."""
+    raw = dict(candidate or {}) if isinstance(candidate, dict) else {}
+    raw_constructs = raw.get("constructs") or []
+    if raw_constructs and all(not isinstance(item, dict) for item in raw_constructs):
+        raw = _legacy_quant_candidate(list(raw_constructs), _clean(raw.get("research_context")))
+        raw_constructs = raw["constructs"]
+
+    constructs: list[dict[str, Any]] = []
+    label_to_id: dict[str, str] = {}
+    for item in raw_constructs[:8]:
+        if not isinstance(item, dict):
+            continue
+        label = _clean(item.get("label") or item.get("name"), 80)
+        if not label or label.casefold() in label_to_id:
+            continue
+        roles = _normalize_roles(item.get("roles") or item.get("role"))
+        if not roles:
+            continue
+        cid = f"C{len(constructs) + 1:03d}"
+        row = {
+            "construct_id": cid,
+            "label": label,
+            "roles": roles,
+            "measurement_model": _clean(item.get("measurement_model") or "reflective", 40).casefold() or "reflective",
+            "measurement_allowed": bool(item.get("measurement_allowed", True)),
+        }
+        constructs.append(row)
+        label_to_id[label.casefold()] = cid
+
+    # Safe compatibility fallback; importantly this is a fallback, not the normal schema.
+    if len(constructs) < 2:
+        raw = _legacy_quant_candidate(["Focal Predictor", "Focal Mediator", "Focal Outcome"], _clean(raw.get("research_context")))
+        return build_quantitative_contract(topic, keywords, raw, source_type=source_type, policy_version=policy_version)
+
+    by_id = {row["construct_id"]: row for row in constructs}
+    def resolve(ref: Any) -> str:
+        text = _clean(ref, 80)
+        if text in by_id:
+            return text
+        return label_to_id.get(text.casefold(), "")
+
+    relations: list[dict[str, Any]] = []
+    rejected: list[dict[str, Any]] = []
+    for item in (raw.get("relations") or [])[:24]:
+        if not isinstance(item, dict):
+            continue
+        rtype = _normalize_quant_relation_type(item.get("type"))
+        source = resolve(item.get("source"))
+        target = resolve(item.get("target"))
+        moderator = resolve(item.get("moderator")) if item.get("moderator") else ""
+        if rtype not in _QUANT_RELATIONS or not source or not target or source == target:
+            rejected.append({"reason": "invalid_or_unsupported_relation", "raw": {k: item.get(k) for k in ("type", "source", "target", "moderator")}})
+            continue
+        if rtype == "moderation" and (not moderator or moderator in {source, target}):
+            rejected.append({"reason": "moderation_requires_distinct_moderator", "raw": {k: item.get(k) for k in ("type", "source", "target", "moderator")}})
+            continue
+        relation = {
+            "relation_id": f"R{len(relations) + 1:03d}",
+            "type": rtype,
+            "source": source,
+            "target": target,
+            "moderator": moderator or None,
+            "analysis_target": bool(item.get("analysis_target", True)),
+            "numbered_hypothesis": bool(item.get("numbered_hypothesis", rtype != "control_effect")),
+        }
+        try:
+            relation["hypothesis_order"] = int(item.get("hypothesis_order")) if item.get("hypothesis_order") is not None else None
+        except Exception:
+            relation["hypothesis_order"] = None
+        relations.append(relation)
+
+    # If the interpreter omitted relations but roles make a parsimonious graph unambiguous,
+    # compile a safe role-derived graph rather than silently returning an empty model.
+    if not relations:
+        predictors = [x["construct_id"] for x in constructs if "predictor" in x["roles"]]
+        mediators = [x["construct_id"] for x in constructs if "mediator" in x["roles"]]
+        outcomes = [x["construct_id"] for x in constructs if "outcome" in x["roles"]]
+        moderators = [x["construct_id"] for x in constructs if "moderator" in x["roles"]]
+        if predictors and outcomes:
+            p, o = predictors[0], outcomes[0]
+            if mediators:
+                m = mediators[0]
+                relations.extend([
+                    {"relation_id":"R001","type":"direct_association","source":p,"target":o,"moderator":None,"analysis_target":True,"numbered_hypothesis":True,"hypothesis_order":1},
+                    {"relation_id":"R002","type":"direct_association","source":p,"target":m,"moderator":None,"analysis_target":True,"numbered_hypothesis":True,"hypothesis_order":2},
+                    {"relation_id":"R003","type":"direct_association","source":m,"target":o,"moderator":None,"analysis_target":True,"numbered_hypothesis":True,"hypothesis_order":3},
+                ])
+            else:
+                relations.append({"relation_id":"R001","type":"direct_association","source":p,"target":o,"moderator":None,"analysis_target":True,"numbered_hypothesis":True,"hypothesis_order":1})
+            if moderators:
+                relations.append({"relation_id":f"R{len(relations)+1:03d}","type":"moderation","source":p,"target":o,"moderator":moderators[0],"analysis_target":True,"numbered_hypothesis":True,"hypothesis_order":len(relations)+1})
+
+    direct_pairs = {(r["source"], r["target"]): r["relation_id"] for r in relations if r["type"] == "direct_association"}
+    indirect_effects: list[dict[str, Any]] = []
+    raw_indirect = raw.get("indirect_effects") or []
+    for item in raw_indirect[:12]:
+        if not isinstance(item, dict):
+            continue
+        path = [resolve(x) for x in (item.get("path") or [])]
+        path = [x for x in path if x]
+        if len(path) < 3 or any((a, b) not in direct_pairs for a, b in zip(path, path[1:])):
+            rejected.append({"reason":"invalid_indirect_path","raw":{"path": item.get("path")}})
+            continue
+        indirect_effects.append({
+            "effect_id": f"I{len(indirect_effects)+1:03d}",
+            "path": path,
+            "kind": _clean(item.get("kind") or ("simple" if len(path)==3 else "serial"), 30).casefold(),
+            "numbered_hypothesis": bool(item.get("numbered_hypothesis", False)),
+        })
+    if not indirect_effects:
+        # Infer valid mediator chains from role identity and existing path edges.
+        predictors = [x["construct_id"] for x in constructs if "predictor" in x["roles"]]
+        mediators = [x["construct_id"] for x in constructs if "mediator" in x["roles"]]
+        outcomes = [x["construct_id"] for x in constructs if "outcome" in x["roles"]]
+        for p in predictors:
+            for m in mediators:
+                for o in outcomes:
+                    if (p, m) in direct_pairs and (m, o) in direct_pairs:
+                        indirect_effects.append({"effect_id":f"I{len(indirect_effects)+1:03d}","path":[p,m,o],"kind":"simple","numbered_hypothesis":False})
+
+    controls = [x["construct_id"] for x in constructs if set(x["roles"]) & {"control", "covariate"}]
+    outcomes = [x["construct_id"] for x in constructs if "outcome" in x["roles"]]
+    if not outcomes:
+        targeted = {r["target"] for r in relations if r["type"] != "control_effect"}
+        outcomes = sorted(targeted)
+
+    analysis_plan = ["measurement_assessment"]
+    if any(r["type"] == "direct_association" for r in relations): analysis_plan.append("direct_association")
+    if indirect_effects: analysis_plan.append("indirect_effect")
+    if any(r["type"] == "moderation" for r in relations): analysis_plan.extend(["interaction_test", "simple_slopes"])
+    analysis_plan.append("robustness")
+
+    # Legacy profile is used only to preserve exact prior H ordering and validators.
+    role_map = {role: [x for x in constructs if role in x["roles"]] for role in ("predictor","mediator","outcome")}
+    legacy_profile = (
+        len(constructs) == 3
+        and all(len(role_map[r]) == 1 for r in role_map)
+        and not any("moderator" in x["roles"] or set(x["roles"]) & {"control","covariate"} for x in constructs)
+        and len([r for r in relations if r["type"] == "direct_association"]) == 3
+    )
+
+    unsupported = _keywords(raw.get("background_concepts") or raw.get("unsupported_seed_concepts") or [])
+    context = _clean(raw.get("research_context") or "respondents with experience relevant to the focal management phenomenon", 240)
+    contract = {
+        "version": STUDY_DESIGN_CONTRACT_VERSION,
+        "mode": "quantitative",
+        "source_type": _clean(source_type, 40) or "system_generated",
+        "source_topic": _clean(topic, 500),
+        "source_keywords": _keywords(keywords),
+        "research_problem": _clean(raw.get("research_problem"), 500),
+        "research_questions": _keywords(raw.get("research_questions") or []),
+        "design_family": "cross_sectional_observational_survey",
+        "evidence_strength": {
+            "causal_identification": False,
+            "allowed_inference": ["association", "estimated_indirect_association", "interaction_association"],
+            "forbidden_inference": ["causal_effect", "causal_mediation_identification"],
+        },
+        "context": {"research_context": context},
+        "quantitative": {
+            "constructs": constructs,
+            "relations": relations,
+            "indirect_effects": indirect_effects,
+            "controls": controls,
+            "outcomes": outcomes,
+            "analysis_plan": analysis_plan,
+            "legacy_three_construct_profile": legacy_profile,
+        },
+        "qualitative": None,
+        "unsupported_seed_concepts": unsupported,
+        "compiler_decisions": rejected,
+        "policy_version": policy_version,
+    }
+    return _finalize(contract)
+
+
+def validate_quantitative_contract(contract: dict[str, Any]) -> list[str]:
+    issues: list[str] = []
+    record = dict(contract or {})
+    if record.get("version") != STUDY_DESIGN_CONTRACT_VERSION or record.get("mode") != "quantitative":
+        return ["invalid_contract_identity"]
+    quant = dict(record.get("quantitative") or {})
+    constructs = [x for x in (quant.get("constructs") or []) if isinstance(x, dict)]
+    ids = {x.get("construct_id") for x in constructs}
+    if len(ids) != len(constructs) or len(constructs) < 2:
+        issues.append("invalid_construct_identity")
+    for r in quant.get("relations") or []:
+        if not isinstance(r, dict):
+            issues.append("invalid_relation_record"); continue
+        if r.get("source") not in ids or r.get("target") not in ids:
+            issues.append("relation_references_unknown_construct")
+        if r.get("type") == "moderation" and r.get("moderator") not in ids:
+            issues.append("moderation_references_unknown_moderator")
+    return list(dict.fromkeys(issues))
+
+
+def compile_quantitative_hypothesis_ledger(contract: dict[str, Any], *, version: str) -> dict[str, Any]:
+    quant = dict((contract or {}).get("quantitative") or {})
+    constructs = {x.get("construct_id"): x for x in (quant.get("constructs") or []) if isinstance(x, dict)}
+    rels = [dict(r) for r in (quant.get("relations") or []) if isinstance(r, dict) and r.get("numbered_hypothesis")]
+    rels.sort(key=lambda r: (r.get("hypothesis_order") is None, r.get("hypothesis_order") or 999, r.get("relation_id") or ""))
+    items: list[dict[str, Any]] = []
+    for r in rels:
+        source = constructs.get(r.get("source"), {}).get("label")
+        target = constructs.get(r.get("target"), {}).get("label")
+        moderator = constructs.get(r.get("moderator"), {}).get("label") if r.get("moderator") else None
+        if not source or not target:
+            continue
+        row = {
+            "hypothesis_id": f"H{len(items)+1}",
+            "predictor": source,
+            "outcome": target,
+            "relation": "moderation" if r.get("type") == "moderation" else "association",
+            "relation_id": r.get("relation_id"),
+        }
+        if moderator:
+            row["moderator"] = moderator
+        items.append(row)
+    indirect_rows = []
+    for effect in quant.get("indirect_effects") or []:
+        if not isinstance(effect, dict): continue
+        labels = [constructs.get(cid, {}).get("label") for cid in effect.get("path") or []]
+        labels = [x for x in labels if x]
+        if len(labels) >= 3:
+            indirect_rows.append({
+                "effect_id": effect.get("effect_id"), "path": labels,
+                "predictor": labels[0], "mediator": " -> ".join(labels[1:-1]), "outcome": labels[-1],
+                "numbered_hypothesis": bool(effect.get("numbered_hypothesis", False)),
+                "relation": "estimated_indirect_association",
+            })
+    ledger = {"version": version, "items": items, "indirect_analyses": indirect_rows}
+    # Exact compatibility key for the historical single-mediator profile.
+    if len(indirect_rows) == 1 and len(indirect_rows[0].get("path") or []) == 3:
+        p = indirect_rows[0]
+        ledger["indirect_analysis"] = {
+            "predictor": p["path"][0], "mediator": p["path"][1], "outcome": p["path"][2],
+            "numbered_hypothesis": p["numbered_hypothesis"], "relation": p["relation"],
+        }
+    return ledger
+
+
+def build_quantitative_front_matter_contract(contract: dict[str, Any], *, version: str) -> dict[str, Any]:
+    quant = dict((contract or {}).get("quantitative") or {})
+    items = [dict(x) for x in (quant.get("constructs") or []) if isinstance(x, dict)]
+    # Preserve legacy role map where a role is singular, while adding an extensible item list.
+    role_map: dict[str, str] = {}
+    for role in ("predictor", "mediator", "moderator", "outcome", "control", "covariate"):
+        values = [x.get("label") for x in items if role in (x.get("roles") or []) and x.get("label")]
+        if len(values) == 1:
+            role_map[role] = values[0]
+    construct_roles = [r for r in ("predictor","mediator","moderator","outcome","control","covariate") if r in role_map]
+    return {
+        "version": version,
+        "policy": "adaptive_study_design_frozen_semantic_authority",
+        "source_topic": str((contract or {}).get("source_topic") or ""),
+        "source_keywords": list((contract or {}).get("source_keywords") or []),
+        "construct_roles": construct_roles,
+        "constructs": role_map,
+        "construct_items": items,
+        "relations": list(quant.get("relations") or []),
+        "unsupported_seed_concepts": list((contract or {}).get("unsupported_seed_concepts") or []),
+        "study_design_contract_sha256": str((contract or {}).get("contract_sha256") or ""),
+    }
+
+
+def build_qualitative_contract(topic: str, keywords: Any, metadata: dict[str, Any], *, policy_version: str = "adaptive-qualitative-v1") -> dict[str, Any]:
+    meta = dict(metadata or {})
+    facts = dict(meta.get("study_facts") or {})
+    sources = []
+    for index, raw in enumerate(facts.get("data_sources") or [], start=1):
+        label = _clean(raw, 120)
+        folded = label.casefold()
+        family = (
+            "interview" if "interview" in folded else
+            "focus_group" if "focus" in folded and "group" in folded else
+            "observation" if "observ" in folded else
+            "log" if "log" in folded else
+            "document" if any(x in folded for x in ("document","archiv","record")) else
+            "digital_trace" if any(x in folded for x in ("digital trace","clickstream","platform trace")) else
+            "field_note" if "field note" in folded else
+            "artifact" if "artifact" in folded else "other"
+        )
+        sources.append({"source_id":f"D{index:03d}","label":label,"family":family,"semantic_descriptor":label if family=="other" else ""})
+    actors = []
+    seen = set()
+    for row in facts.get("data_source_roles") or []:
+        if not isinstance(row, dict): continue
+        label = _clean(row.get("actor_role"), 100)
+        if label and label.casefold() not in seen:
+            actors.append({"actor_id":f"A{len(actors)+1:03d}","label":label}); seen.add(label.casefold())
+    themes = []
+    for item in meta.get("themes") or []:
+        if not isinstance(item, dict): continue
+        themes.append({
+            "theme_id": _clean(item.get("theme_id"), 40), "title": _clean(item.get("title"), 160),
+            "stages": [_clean(x,80) for x in (item.get("stages") or []) if _clean(x,80)],
+            "has_deviant_evidence": bool(item.get("has_deviant_evidence")),
+        })
+    design = _clean(facts.get("analytic_design") or facts.get("analysis_method"), 120)
+    temporal = _clean(facts.get("temporal_structure"), 160)
+    case_structure = _clean(facts.get("case_structure"), 160)
+    contract = {
+        "version": STUDY_DESIGN_CONTRACT_VERSION,
+        "mode": "qualitative",
+        "source_type": _clean(meta.get("source_type"),40),
+        "source_topic": _clean(topic,500),
+        "source_keywords": _keywords(keywords),
+        "research_problem": "",
+        "research_questions": [],
+        "design_family": design or "Iterative thematic/process analysis",
+        "evidence_strength": {"causal_identification":False,"allowed_inference":["theme","process_pattern","interpretive_mechanism"],"forbidden_inference":["causal_effect"]},
+        "context": {"research_setting": _clean(facts.get("research_setting"),240)},
+        "quantitative": None,
+        "qualitative": {
+            "phenomenon": _clean(topic,300),
+            "organizational_puzzle": "",
+            "research_questions": [],
+            "design_tradition": design or "Iterative thematic/process analysis",
+            "units_of_analysis": [{"unit_id":"U001","label":case_structure or "study participants/organizational setting","level":"other"}],
+            "actor_groups": actors,
+            "data_sources": sources,
+            "sampling": {"participant_count": facts.get("participant_count"), "interview_count": facts.get("interview_count")},
+            "temporal_scope": {"description": temporal or "Not specified"},
+            "analysis_strategy": [design] if design else [],
+            "themes": themes,
+            "process_model": [],
+            "trustworthiness_procedures": list(facts.get("trustworthiness_procedures") or []),
+            "ethics_procedures": list(facts.get("ethics_procedures") or []),
+            "reflexivity_procedures": list(facts.get("reflexivity_procedures") or []),
+        },
+        "unsupported_seed_concepts": [],
+        "compiler_decisions": [],
+        "policy_version": policy_version,
+    }
+    return _finalize(contract)
+
+
+def _study_owned_sentence(sentence: str) -> bool:
+    text = str(sentence or "")
+    if re.search(r"\bCITE-\d+\b", text, flags=re.I): return False
+    if re.search(r"\((?=[^()\n]*(?:19|20)\d{2})[^()\n]+\)", text): return False
+    if re.search(r"(?i)\b(?:prior|previous|existing|earlier)\s+(?:research|stud(?:y|ies)|literature)\b", text): return False
+    return bool(re.search(r"(?i)\b(?:we|our\s+study|this\s+study|the\s+present\s+study|the\s+study|our\s+analysis|our\s+results|our\s+findings|we\s+(?:examine|investigate|test|analy[sz]e|explore|study))\b", text))
+
+
+def inspect_study_design_scope(text: str, contract: dict[str, Any], *, section_name: str = "") -> list[dict[str, Any]]:
+    """High-confidence whole-manuscript scope guard for both modes."""
+    value = str(text or "")
+    rows: list[dict[str, Any]] = []
+    mode = str((contract or {}).get("mode") or "")
+    sentences = [s.strip() for s in re.split(r"(?<=[.!?])\s+|\n+", value) if s.strip()]
+    if mode == "quantitative":
+        quant = dict((contract or {}).get("quantitative") or {})
+        constructs = [x for x in (quant.get("constructs") or []) if isinstance(x, dict)]
+        allowed_labels = {str(x.get("label") or "").casefold() for x in constructs if x.get("label")}
+        has_moderation = any(r.get("type") == "moderation" for r in (quant.get("relations") or []) if isinstance(r, dict))
+        has_indirect = bool(quant.get("indirect_effects"))
+        unsupported = [str(x) for x in (contract or {}).get("unsupported_seed_concepts") or [] if str(x).strip()]
+        for sentence in sentences:
+            if not _study_owned_sentence(sentence): continue
+            folded = sentence.casefold()
+            if re.search(r"\bmoderat(?:e|es|ed|ing|or|ion)\b|\binteraction\s+(?:effect|term)\b", folded) and not has_moderation:
+                rows.append({"type":"unsupported_moderation_claim","preview":sentence[:260],"reason":"The frozen study-design graph contains no moderation/interaction analysis target."})
+            if re.search(r"\bmediat(?:e|es|ed|ing|or|ion)\b|\bindirect\s+(?:effect|association)\b", folded) and not has_indirect:
+                rows.append({"type":"unsupported_mediation_claim","preview":sentence[:260],"reason":"The frozen study-design graph contains no indirect-effect analysis target."})
+            for concept in unsupported:
+                if concept.casefold() in folded and re.search(r"(?i)\b(?:examine|investigate|test|model|measure|assess|estimate)\b", sentence):
+                    rows.append({"type":"unsupported_focal_construct","construct":concept,"preview":sentence[:260],"reason":"The concept is background-only under the frozen study-design contract."})
+    elif mode == "qualitative":
+        qual = dict((contract or {}).get("qualitative") or {})
+        design = str(qual.get("design_tradition") or "").casefold()
+        temporal = str(dict(qual.get("temporal_scope") or {}).get("description") or "").casefold()
+        families = {str(x.get("family") or "") for x in (qual.get("data_sources") or []) if isinstance(x, dict)}
+        unit_label = " ".join(str(x.get("label") or "") for x in (qual.get("units_of_analysis") or []) if isinstance(x, dict)).casefold()
+        for sentence in sentences:
+            if not _study_owned_sentence(sentence): continue
+            folded = sentence.casefold()
+            if "longitudinal" in folded and "longitudinal" not in temporal and "longitudinal" not in design:
+                rows.append({"type":"unsupported_temporal_scope","preview":sentence[:260],"reason":"The frozen qualitative design does not establish longitudinal evidence."})
+            if "ethnograph" in folded and "ethnograph" not in design:
+                rows.append({"type":"unsupported_design_tradition","preview":sentence[:260],"reason":"Ethnography is not the frozen qualitative design tradition."})
+            if "grounded theory" in folded and "grounded theory" not in design:
+                rows.append({"type":"unsupported_design_tradition","preview":sentence[:260],"reason":"Grounded theory is not the frozen qualitative design tradition."})
+            if "triangulat" in folded and len({x for x in families if x}) < 2:
+                rows.append({"type":"unsupported_data_source_claim","preview":sentence[:260],"reason":"Triangulation requires at least two frozen evidence-source families."})
+            if re.search(r"\b(?:multi|multiple)[- ]case\b|\bcomparative case\b", folded) and not any(x in (design + " " + unit_label) for x in ("comparative", "multiple", "multi-case")):
+                rows.append({"type":"unsupported_unit_of_analysis","preview":sentence[:260],"reason":"The frozen qualitative contract does not establish a multi-case comparison."})
+    return rows[:8]