A5-R3-R7-R2C-R5 user research intent authority scoped deployment

core/generation_runner.py

@@ -32,6 +32,7 @@ from core.production_guard import build_validated_package, replace_abstract
 from core.qualitative_engine import get_qualitative_engine
 from core.quantitative_engine import get_quantitative_engine
 from core.rag_engine import initialize_rag
+from core.study_design_contract import build_user_research_intent
 from utils.file_parser import extract_text_from_files
 
 ProgressCallback = Callable[..., None]
@@ -1796,10 +1797,19 @@ class GenerationRunner:
 
     @staticmethod
     def _canonical_quantitative_front_matter_keywords(mode: str, candidate_keywords, data_metadata: dict):
-        """Normalize model-generated keywords without replacing semantic context."""
+        """Return user-authoritative keywords when available; otherwise normalize a legacy candidate.
+
+        The name is retained for checkpoint compatibility, but R7-R2C-R5 applies
+        the same authority rule to quantitative and qualitative manuscripts.
+        """
+        meta = dict(data_metadata or {})
+        contract = dict(meta.get("study_design_contract") or {})
+        intent = dict(contract.get("user_research_intent") or meta.get("user_research_intent") or {})
+        authoritative = list(intent.get("source_keywords") or contract.get("source_keywords") or [])
+        source = authoritative if authoritative else list(candidate_keywords or [])
         values = []
         seen = set()
-        for item in candidate_keywords or []:
+        for item in source:
             value = re.sub(r"\s+", " ", str(item or "")).strip()
             key = value.casefold()
             if value and key not in seen:
@@ -1842,6 +1852,85 @@ class GenerationRunner:
             return value[:title_match.end()] + "\n\n" + block + value[title_match.end():].lstrip("\n")
         return block + value.lstrip("\n")
 
+
+    @staticmethod
+    def _target_keyword_heading_aliases(target_lang: str) -> list[str]:
+        code = str(target_lang or '').strip().casefold().replace('_', '-')
+        if code.startswith('zh'):
+            return ['关键词', '关键字']
+        if code.startswith('ru'):
+            return ['Ключевые слова']
+        if code.startswith('kk') or code.startswith('kz'):
+            return ['Түйін сөздер', 'Кілт сөздер']
+        return ['Keywords']
+
+    @classmethod
+    def _rebind_target_user_intent_front_matter(cls, manuscript: str, target_lang: str, data_metadata: dict, glossary: dict):
+        """Preserve user keyword identity and explicit relation type in target front matter."""
+        value = str(manuscript or '')
+        meta = dict(data_metadata or {})
+        contract = dict(meta.get('study_design_contract') or {})
+        intent = dict(contract.get('user_research_intent') or meta.get('user_research_intent') or {})
+        source_keywords = list(intent.get('source_keywords') or contract.get('source_keywords') or [])
+        mapping = dict(glossary or {})
+        diagnostics = []
+
+        if source_keywords:
+            target_keywords = [re.sub(r'\s+', ' ', str(mapping.get(src) or '')).strip() for src in source_keywords]
+            if all(target_keywords):
+                aliases = cls._target_keyword_heading_aliases(target_lang)
+                heading_re = re.compile(r'(?mi)^##\s+(?:' + '|'.join(re.escape(x) for x in aliases) + r')\s*$')
+                match = heading_re.search(value)
+                if match:
+                    next_heading = re.search(r'(?m)^#{1,3}\s+', value[match.end():])
+                    end = match.end() + (next_heading.start() if next_heading else len(value[match.end():]))
+                    rendered = ', '.join(target_keywords)
+                    value = value[:match.end()] + '\n' + rendered + '\n\n' + value[end:].lstrip('\n')
+                    diagnostics.append({'type':'target_user_keywords_rebound','count':len(target_keywords)})
+
+        explicit = set(intent.get('explicit_relation_types') or [])
+        if explicit:
+            title_match = re.search(r'(?m)^#\s+(.+?)\s*$', value)
+            if title_match:
+                title = title_match.group(1).strip()
+                target_intent = build_user_research_intent(title, '', mode=str(contract.get('mode') or ''))
+                missing = explicit - set(target_intent.get('explicit_relation_types') or [])
+                if missing:
+                    quant = dict(contract.get('quantitative') or {})
+                    construct_rows = [row for row in (quant.get('constructs') or []) if isinstance(row, dict)]
+                    role_label = {}
+                    for role in ('mediator', 'moderator'):
+                        values = [str(row.get('label') or '') for row in construct_rows if role in (row.get('roles') or []) and row.get('label')]
+                        if len(values) == 1:
+                            role_label[role] = re.sub(r'\s+', ' ', str(mapping.get(values[0]) or values[0])).strip()
+                    code = str(target_lang or '').strip().casefold().replace('_', '-')
+                    suffixes = []
+                    if 'moderation' in missing:
+                        label = role_label.get('moderator') or ''
+                        if code.startswith('zh'):
+                            suffixes.append(f'调节变量：{label}' if label else '调节作用')
+                        elif code.startswith('ru'):
+                            suffixes.append(f'модератор: {label}' if label else 'модерация')
+                        elif code.startswith('kk') or code.startswith('kz'):
+                            suffixes.append(f'модератор: {label}' if label else 'модераторлық рөл')
+                        else:
+                            suffixes.append(f'moderator: {label}' if label else 'moderation')
+                    if missing.intersection({'mediation', 'serial_mediation', 'parallel_mediation'}):
+                        label = role_label.get('mediator') or ''
+                        if code.startswith('zh'):
+                            suffixes.append(f'中介变量：{label}' if label else '中介作用')
+                        elif code.startswith('ru'):
+                            suffixes.append(f'медиатор: {label}' if label else 'медиация')
+                        elif code.startswith('kk') or code.startswith('kz'):
+                            suffixes.append(f'медиатор: {label}' if label else 'медиаторлық рөл')
+                        else:
+                            suffixes.append(f'mediator: {label}' if label else 'mediation')
+                    if suffixes:
+                        repaired_title = title.rstrip(' .;:') + ' (' + '; '.join(suffixes) + ')'
+                        value = value[:title_match.start(1)] + repaired_title + value[title_match.end(1):]
+                        diagnostics.append({'type':'target_user_relation_title_rebound','missing':sorted(missing)})
+        return value, diagnostics
+
     @staticmethod
     def _section_is_complete(tasks, completed_display_names, section_key: str) -> bool:
         relevant = [
@@ -1890,10 +1979,19 @@ class GenerationRunner:
             items.append({'source': term, 'kind': str(kind or 'concept'), 'policy': normalized_policy})
 
         meta = dict(data_metadata or {})
+        contract = dict(meta.get('study_design_contract') or {})
+        user_intent = dict(contract.get('user_research_intent') or meta.get('user_research_intent') or {})
         if str(mode or '') == 'amj':
-            constructs = meta.get('constructs') if isinstance(meta.get('constructs'), dict) else {}
-            for role in ('predictor', 'mediator', 'outcome'):
-                add(constructs.get(role), f'quantitative_{role}', 'STRICT_CONSTRUCT')
+            construct_items = [row for row in (meta.get('construct_items') or []) if isinstance(row, dict)]
+            if construct_items:
+                for row in construct_items:
+                    roles = list(row.get('roles') or [])
+                    role_label = '_'.join(str(role) for role in roles if str(role)) or 'construct'
+                    add(row.get('label'), f'quantitative_{role_label}', 'STRICT_CONSTRUCT')
+            else:
+                constructs = meta.get('constructs') if isinstance(meta.get('constructs'), dict) else {}
+                for role in ('predictor', 'mediator', 'moderator', 'outcome', 'control', 'covariate'):
+                    add(constructs.get(role), f'quantitative_{role}', 'STRICT_CONSTRUCT')
         else:
             for theme in (meta.get('themes') or []):
                 if not isinstance(theme, dict):
@@ -1911,14 +2009,15 @@ class GenerationRunner:
             for concept in (task.get('concepts') or []):
                 add(concept, 'outline_concept', 'FLEXIBLE_TERM')
 
-        # User keywords remain planning seeds, not canonical identities. Use
-        # them only as a fallback when frozen empirical/qualitative metadata and
-        # outline concepts provide no concept labels at all.
-        if not items:
-            for seed in re.split(r'[,;\n|]+', str(keywords or '')):
-                seed = re.sub(r'\s+', ' ', seed).strip()
-                if 1 <= len(seed.split()) <= 6:
-                    add(seed, 'keyword_fallback', 'FLEXIBLE_TERM')
+        # R7-R2C-R5: user keywords are research-intent identities, not expendable
+        # planning seeds.  Freeze them for multilingual inheritance while still
+        # distinguishing them from empirical construct roles.
+        intent_keywords = list(user_intent.get('source_keywords') or [])
+        if not intent_keywords:
+            intent_keywords = [seed.strip() for seed in re.split(r'[,;\n|]+', str(keywords or '')) if seed.strip()]
+        for seed in intent_keywords:
+            if 1 <= len(str(seed).split()) <= 10:
+                add(seed, 'user_intent_keyword', 'STRICT_CONSTRUCT')
 
         # Capture explicit lower-case hyphenated academic expressions only when
         # they recur or carry a research-concept cue. This closes the
@@ -1937,7 +2036,13 @@ class GenerationRunner:
             if counts.get(term.casefold(), 0) >= 2 or parts.intersection(concept_cues):
                 add(term, 'logic_hyphenated_concept', 'FLEXIBLE_TERM')
 
-        return {'version': 'concept-ledger-v2', 'items': items}
+        return {
+            'version': 'concept-ledger-v2',
+            'items': items,
+            'explicit_relation_types': list(user_intent.get('explicit_relation_types') or []),
+            'source_topic': str(user_intent.get('source_topic') or contract.get('source_topic') or ''),
+            'source_keywords': list(user_intent.get('source_keywords') or contract.get('source_keywords') or []),
+        }
 
     @staticmethod
     def _canonical_concept_writer_directive(ledger: dict) -> str:
@@ -1946,12 +2051,21 @@ class GenerationRunner:
         if not terms:
             return ''
         rendered = '; '.join(terms[:40])
+        relation_types = [str(item).strip() for item in (dict(ledger or {}).get('explicit_relation_types') or []) if str(item).strip()]
+        relation_directive = ''
+        if relation_types:
+            relation_directive = (
+                ' USER RESEARCH INTENT: Preserve these explicitly requested study relations throughout drafting and target-language translation: '
+                + ', '.join(relation_types)
+                + '. Do not downgrade a mediator/moderator or other explicit relation into a generic contextual term merely because direct prior-literature support is sparse. Prior-literature claims still require real citation evidence; novel study logic may be stated cautiously as author inference without inventing a source.'
+            )
         return (
             'CANONICAL CONCEPT IDENTITY: The following research labels are frozen task-level concepts: '
             + rendered
             + '. When referring to the same construct or concept, preserve its canonical English label; '
               'do not silently replace it with a near-synonym that changes concept identity. Ordinary grammatical '
               'articles, possessives, and surrounding syntax remain free. Distinct listed labels are not synonyms unless the evidence explicitly says so.'
+            + relation_directive
         )
 
     @staticmethod
@@ -3443,7 +3557,9 @@ class GenerationRunner:
 
             self._progress(14, "outline_planning")
             plan = agent.generate_outline(topic, keywords, mode=mode, data_metadata=data_metadata)
-            en_topic = str(plan.get("en_topic", topic))
+            # User title is reader-facing authority.  Outline planning may adapt
+            # structure but may not silently rename the study.
+            en_topic = str(source_topic or topic).strip()
             structure = plan.get("structure", [])
             full_draft_en = (
                 f"# {en_topic}\n\n## Abstract\n{working_abstract}\n\n"
@@ -5190,7 +5306,7 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     "examples":final_front_fact_rows[:3],
                 })
             state.pop("final_abstract_retry_exhausted", None)
-            final_title = str(topic or "").strip()
+            final_title = str(source_topic or topic or "").strip()
             if mode == "amj" and str(data_metadata.get("source_type") or "") == "system_generated":
                 semantic_contract = dict(data_metadata.get("front_matter_contract") or {})
                 reconciled_front = agent.reconcile_quantitative_reader_front_matter(
@@ -5456,6 +5572,14 @@ The previous attempt for this same manuscript unit was rejected by validation. R
             full_draft_translated = _normalize_target_manuscript_typography(
                 full_draft_translated, target_lang
             )
+            full_draft_translated, intent_front_diagnostics = self._rebind_target_user_intent_front_matter(
+                full_draft_translated, target_lang, data_metadata, translation_glossary
+            )
+            for diagnostic in intent_front_diagnostics:
+                self._record_adaptive_completion_residual(
+                    quality_notes, [diagnostic], section_name='Target-language front matter',
+                    stage='user_research_intent_translation_recovery'
+                )
             if mode == "asq":
                 appendix_manifest = qual_engine.build_appendix_manifest(data_metadata)
                 try:

core/pipeline_agent.py

@@ -1838,6 +1838,27 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
         return issues
 
     def reconcile_quantitative_reader_front_matter(self, *, topic, keywords, contract, manuscript=""):
+        """Preserve R5 user intent while retaining legacy contract reconciliation.
+
+        Contracts that carry ``user_research_intent`` are authoritative: the
+        finalizer must not repair an upstream design mismatch by silently
+        renaming the user study.  Older contracts without that explicit R5
+        authority retain the pre-R5 bounded reconciliation path so historical
+        checkpoints and compatibility behavior remain stable.
+        """
+        record = dict(contract or {})
+        intent = dict(record.get("user_research_intent") or {})
+        if intent:
+            source_topic = str(intent.get("source_topic") or record.get("source_topic") or topic or "").strip()
+            source_keywords = list(intent.get("source_keywords") or record.get("source_keywords") or [])
+            if not source_keywords:
+                source_keywords = [item.strip() for item in re.split(r"[,;\n|]+", str(keywords or "")) if item.strip()]
+            return {
+                "title": source_topic,
+                "keywords": [re.sub(r"\s+", " ", str(item or "")).strip() for item in source_keywords if str(item or "").strip()],
+            }
+
+        # Legacy compatibility: no R5 user-intent authority was frozen.
         """Adaptively reconcile final AMJ title/keywords to frozen evidence.
 
         Original user topic/keyword seeds are retained as research context.  The
@@ -1954,6 +1975,7 @@ A contextual/domain term is allowed. Mark INVALID only when the proposal makes a
         logging.warning("Adaptive front-matter reconciliation exhausted; using deterministic degraded fallback instead of failing manuscript generation.")
         return fallback
 
+
     def generate_working_front_matter(
         self, topic, keywords, research_logic_contract, synthesis_result, data_evidence='', mode='amj', validation_feedback='', hypothesis_contract=''
     ):
@@ -1964,9 +1986,9 @@ A contextual/domain term is allowed. Mark INVALID only when the proposal makes a
         prompt = f"""Create WORKING front matter for an empirical {str(mode).upper()}-oriented manuscript before body drafting.
 Return JSON only: {{\"abstract\":\"...\",\"keywords\":[\"...\"]}}.
 Aim for approximately 120-180 English words, have no citations, and function only as a continuity anchor. Word count is guidance, not a failure condition; prioritize Conciseness, Accuracy, and Rigor over mechanical compliance. It may state the research problem, theoretical tension, focal concepts, question/hypothesis architecture, design or method class, and intended contribution domain. It MUST NOT state coefficients, p-values, fit statistics, sample statistics unless already indispensable to define the design, supported/unsupported hypotheses, result direction, substantive findings, or a conclusion that has not yet been frozen. Do not predict findings.
-Return 4-6 concise academic keywords. User keywords are seeds, not an authoritative final list.
+Preserve the user keyword identities and research scope. Do not silently replace, delete, or reinterpret an explicit construct/relation because direct literature support is sparse. The user-supplied keyword list is authoritative for reader-facing keyword identity; wording changes belong only to target-language translation.
 TOPIC: {topic}
-USER KEYWORD SEEDS: {keywords}
+USER KEYWORDS: {keywords}
 <PROVISIONAL_RESEARCH_LOGIC>
 {logic}
 </PROVISIONAL_RESEARCH_LOGIC>
@@ -2000,9 +2022,9 @@ If the immutable hypothesis contract is non-empty, preserve its H-identifiers an
         prompt = f"""Write FINAL English front matter for this completed {str(mode).upper()} manuscript.
 Return JSON only: {{\"abstract\":\"...\",\"keywords\":[\"...\"]}}.
 Aim for a concise Abstract of roughly 120-200 words, use no citations, and accurately state the research problem, validated design, main frozen finding(s), and contribution. Word count is guidance, not a failure condition; prioritize Conciseness, Accuracy, and Rigor. Include only the minimum numerical evidence required; do not list coefficients, p-values, and fit statistics unless one value is indispensable. Do not infer beyond the frozen evidence or completed manuscript.
-Return 4-6 concise final academic keywords derived from the completed manuscript. User keyword seeds are advisory only.
+Preserve the user keyword identities and research scope. Do not silently replace, delete, or reinterpret an explicit construct/relation because direct literature support is sparse. The user-supplied keyword list is authoritative for reader-facing keyword identity; wording changes belong only to target-language translation.
 TOPIC: {topic}
-USER KEYWORD SEEDS: {keywords}
+USER KEYWORDS: {keywords}
 <FROZEN_RESEARCH_EVIDENCE>
 {evidence}
 </FROZEN_RESEARCH_EVIDENCE>

core/qualitative_engine.py

@@ -9,7 +9,11 @@ import logging
 import re
 
 from config import APP_CONFIG
-from core.study_design_contract import STUDY_DESIGN_CONTRACT_VERSION, build_qualitative_contract
+from core.study_design_contract import (
+    STUDY_DESIGN_CONTRACT_VERSION,
+    build_qualitative_contract,
+    build_user_research_intent,
+)
 
 
 class QualitativeEngine:
@@ -568,7 +572,28 @@ class QualitativeEngine:
             )
         return "\n".join(lines).strip()
 
+    @staticmethod
+    def _intent_data_source_family(value):
+        folded = re.sub(r"\s+", " ", str(value or "")).strip().casefold()
+        if re.search(r"interview|访谈|интервью|сұхбат", folded):
+            return "interview"
+        if re.search(r"focus.{0,4}(?:group|групп|топ)|焦点小组", folded):
+            return "focus_group"
+        if re.search(r"observ|观察|наблюден|бақылау", folded):
+            return "observation"
+        if re.search(r"archiv|document|档案|文档|文件|архив|документ|мұрағат|құжат", folded):
+            return "document"
+        if re.search(r"field.{0,4}note|田野笔记|полев.{0,6}замет|далалық.{0,6}жазб", folded):
+            return "field_note"
+        if re.search(r"digital.{0,4}trace|数字痕迹|цифров.{0,6}след|цифрлық.{0,6}із", folded):
+            return "digital_trace"
+        if re.search(r"\blog(?:s)?\b|日志|\bлог\b", folded):
+            return "log"
+        return "other"
+
     def _build_request(self, topic, keywords, raw_interviews):
+        user_intent = build_user_research_intent(topic, keywords, mode="qualitative")
+        intent_block = json.dumps(user_intent, ensure_ascii=False)
         schema = """Return ONLY one valid JSON object with this schema:
 {
   "study_facts": {
@@ -601,9 +626,11 @@ Allowed analytic designs: Process analysis; Comparative case analysis; Grounded
         if raw_interviews and len(raw_interviews) > 100:
             return f"""Topic: {topic}
 Keywords: {keywords}
+<USER_RESEARCH_INTENT>{intent_block}</USER_RESEARCH_INTENT>
 Analyze the supplied interview or field material for an ASQ-oriented qualitative manuscript.
 
 Rules:
+0. USER RESEARCH INTENT AUTHORITY: Preserve the user's research phenomenon, explicitly requested qualitative design, data-source family, temporal scope, actor/population, and context when the supplied material supports them. Never silently substitute a different method or phenomenon merely because literature or a generic template is easier. If the supplied material genuinely conflicts with the requested design, preserve evidence truth rather than inventing support; the deterministic validator will surface the conflict.
 1. Treat the supplied material as the only source of study facts. Use null, empty arrays, or "Not specified" for any setting, sample, count, date, case, actor/data-source relation, trustworthiness procedure, ethics procedure, reflexivity procedure, or data source not explicitly supported by the material.
 2. Extract 8-15 representative quotations VERBATIM. Never repair, paraphrase, translate, combine, or invent a quotation.
 3. Choose the analytic design that best fits the actual material. Do NOT default to Gioia. Use Gioia-style coding only if the evidence structure genuinely warrants it.
@@ -619,9 +646,11 @@ Rules:
 
         return f"""Topic: {topic}
 Keywords: {keywords}
+<USER_RESEARCH_INTENT>{intent_block}</USER_RESEARCH_INTENT>
 Construct one internally coherent qualitative evidence packet for a complete ASQ-oriented manuscript because no interview material was supplied.
 
 Rules:
+0. USER RESEARCH INTENT AUTHORITY: The topic/keywords define what the study is asked to investigate. If they explicitly request an analytic design (for example grounded theory, thematic analysis, comparative case, process, longitudinal, or Gioia) or a data-source family (for example interviews, focus groups, observation, documents), preserve that request in the generated evidence packet. Sparse literature is not a reason to substitute another method.
 1. Choose one generic organizational research setting directly relevant to the topic and keep it fixed. Do not name a real organization, institution, or identifiable person.
 2. Freeze one participant count, interview count, data-source set, actor-to-data-source relation map, analytic design, temporal/case structure, and analytic logic. Keep them internally consistent. Every actor/data-source relationship later stated in the paper must be represented in data_source_roles.
 3. Produce exactly 12 natural representative quotations distributed across at least three higher-order themes. Quotations must sound like distinct participants rather than repeated template variants.
@@ -635,7 +664,7 @@ Rules:
 {schema}
 """
 
-    def _qualitative_packet_validation_error(self, data, *, has_user_data):
+    def _qualitative_packet_validation_error(self, data, *, has_user_data, user_intent=None):
         """Return a bounded local validation reason, or an empty string when safe.
 
         This pre-freeze validator mirrors the existing fatal packet gates so a
@@ -688,6 +717,19 @@ Rules:
                 if not (facts.get(key) or []):
                     return f"generated_packet_missing_{key}"
 
+        intent = dict(user_intent or {})
+        requested_design = str(intent.get("qualitative_design") or "").strip()
+        if requested_design:
+            observed_design = self._normalize_analytic_design(facts.get("analytic_design") or facts.get("analysis_method"))
+            if observed_design.casefold() != requested_design.casefold():
+                return f"user_intent_analytic_design_mismatch:{observed_design}->{requested_design}"
+        requested_sources = set(intent.get("qualitative_data_source_families") or [])
+        if requested_sources:
+            observed_sources = {self._intent_data_source_family(item) for item in (facts.get("data_sources") or [])}
+            missing_sources = sorted(requested_sources - observed_sources)
+            if missing_sources:
+                return "user_intent_data_source_mismatch:" + ",".join(missing_sources)
+
         seen_quotes = set()
         theme_keys = set()
         retained = 0
@@ -721,6 +763,7 @@ Rules:
             "You are a rigorous qualitative research analyst. "
             "Evidence fidelity has priority over narrative completeness."
         )
+        user_intent = build_user_research_intent(topic, keywords, mode="qualitative")
         base_prompt = self._build_request(topic, keywords, raw_interviews)
         last_reason = "packet_not_json_object"
 
@@ -769,6 +812,7 @@ Rules:
             last_reason = self._qualitative_packet_validation_error(
                 data,
                 has_user_data=has_user_data,
+                user_intent=user_intent,
             )
             if not last_reason:
                 return data
@@ -983,6 +1027,7 @@ Rules:
             },
             "evidence": clean_evidence,
             "themes": themes,
+            "user_research_intent": build_user_research_intent(topic, keywords, mode="qualitative"),
         }
         metadata["study_design_contract"] = build_qualitative_contract(topic, keywords, metadata)
         metadata["semantic_claim_frame"] = self.build_qualitative_semantic_claim_frame(metadata)

core/quantitative_engine.py

@@ -19,6 +19,7 @@ from core.study_design_contract import (
     validate_quantitative_contract,
     compile_quantitative_hypothesis_ledger,
     build_quantitative_front_matter_contract,
+    build_user_research_intent,
 )
 
 
@@ -473,6 +474,7 @@ class QuantitativeEngine:
         The existing ``structured_json`` logical node is intentionally reused so
         Standard/Flash and Flagship/Pro ModelPolicy routing remains unchanged.
         """
+        user_intent = build_user_research_intent(topic, keywords, mode="quantitative")
         fallback_candidate = {
             "constructs": [
                 {"label":"Focal Predictor","roles":["predictor"]},
@@ -500,8 +502,9 @@ Return ONLY valid JSON with this schema:
   "background_concepts":["topic concept that is contextual rather than measured/modelled"]
 }}
 Rules:
-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.
+0. USER RESEARCH INTENT AUTHORITY: The user's topic and keywords are deliberate research-design inputs. Never delete, replace, or downgrade an explicitly requested focal construct, mediation, moderation, serial/parallel mediation, control, population, or research context merely because direct literature support may be sparse. Literature evidence constrains claims about PRIOR research; it does not redefine what this study is asked to investigate.
+1. Use 2-8 distinct constructs only when the topic actually requires them; do NOT force a mediator or moderator that the user did not request.
+2. Preserve explicit mediation, moderation, serial/parallel mediation, multiple antecedents/outcomes, and controls when the topic clearly requests them. When the requested relation is novel or direct prior evidence is sparse, keep the relation in the study graph and let later hypothesis development use cautious author inference rather than inventing citation support.
 3. A moderation relation means source × moderator is associated with target.
 4. Keep the graph parsimonious and acyclic for ordinary cross-sectional regression interpretation.
 5. Use concise construct labels suitable for a journal article.
@@ -526,8 +529,12 @@ Keywords: {keywords}"""
             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)
+                logging.warning("Adaptive quantitative design validation requires intent-preserving fallback: %s", issues[:3])
+                fallback = build_quantitative_contract(topic, keywords, fallback_candidate)
+                fallback_issues = validate_quantitative_contract(fallback)
+                if fallback_issues:
+                    raise RuntimeError(f"quantitative_user_intent_unresolved:{fallback_issues[:3]}")
+                return fallback
             context = str(dict(contract.get("context") or {}).get("research_context") or "")
             if re.search(r"\b(university|company|corporation|inc\.?|ltd\.?|country|china|usa|united states|europe|asia)\b", context, re.I):
                 fallback_context = "respondents with experience relevant to the focal management phenomenon"

core/study_design_contract.py

@@ -40,6 +40,290 @@ def _keywords(value: Any) -> list[str]:
     return out[:16]
 
 
+
+USER_RESEARCH_INTENT_VERSION = "smartpaper-user-research-intent-v1"
+
+
+def _verbatim_keywords(value: Any) -> list[str]:
+    """Preserve user keyword identity/order while trimming separator whitespace only."""
+    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 = re.sub(r"\s+", " ", str(item or "")).strip()
+        key = text.casefold()
+        if text and key not in seen:
+            out.append(text)
+            seen.add(key)
+    return out[:24]
+
+
+_MODERATION_CUES = (
+    r"\bmoderating\s+role\b", r"\bmoderation\b", r"\bmoderator\b", r"\bmoderates?\b",
+    r"调节作用", r"调节效应", r"调节变量", r"调节关系",
+    r"модерирующ", r"модератор", r"модерац",
+    r"модераторлық", r"модерациялық", r"модератор ретінде",
+)
+_MEDIATION_CUES = (
+    r"\bmediating\s+role\b", r"\bmediation\b", r"\bmediator\b", r"\bmediates?\b",
+    r"中介作用", r"中介效应", r"中介变量", r"链式中介", r"并行中介",
+    r"медиирующ", r"медиатор", r"медиац",
+    r"медиаторлық", r"медиациялық", r"медиатор ретінде",
+)
+_SERIAL_MEDIATION_CUES = (
+    r"\bserial\s+mediat", r"\bsequential\s+mediat", r"链式中介", r"序列中介",
+    r"последовательн.{0,16}медиац", r"тізбекті.{0,16}медиац",
+)
+_PARALLEL_MEDIATION_CUES = (
+    r"\bparallel\s+mediat", r"并行中介", r"параллельн.{0,16}медиац", r"параллель.{0,16}медиац",
+)
+
+_QUAL_DESIGN_PATTERNS = (
+    ("Grounded theory", (r"\bgrounded\s+theory\b", r"扎根理论", r"обоснованн.{0,8}теори", r"негізделген\s+теори")),
+    ("Gioia-style inductive coding", (r"\bgioia\b", r"Gioia方法", r"Gioia 方法", r"джойя", r"гиоиа")),
+    ("Comparative case analysis", (r"\bcomparative\s+(?:case|cases)\b", r"比较案例", r"比较个案", r"сравнительн.{0,10}(?:кейс|случа)", r"салыстырмалы.{0,10}(?:кейс|жағдай)")),
+    ("Longitudinal qualitative analysis", (r"\blongitudinal\b", r"纵向(?:定性|质性)?", r"продольн", r"лонгитюд", r"бойлық")),
+    ("Process analysis", (r"\bprocess\s+analysis\b", r"过程分析", r"процессн.{0,8}анализ", r"үдерістік.{0,8}талдау", r"процестік.{0,8}талдау")),
+    ("Iterative thematic/process analysis", (r"\bthematic\s+analysis\b", r"主题分析", r"тематическ.{0,8}анализ", r"тақырыптық.{0,8}талдау")),
+)
+
+_QUAL_SOURCE_PATTERNS = (
+    ("interview", (r"\bsemi[- ]structured\s+interview", r"\binterviews?\b", r"半结构式访谈", r"半结构访谈", r"访谈", r"полуструктурированн.{0,8}интервью", r"интервью", r"жартылай құрылымданған.{0,8}сұхбат", r"сұхбат")),
+    ("focus_group", (r"\bfocus\s+groups?\b", r"焦点小组", r"фокус[- ]?групп", r"фокус[- ]?топ")),
+    ("observation", (r"\bobservation(?:s)?\b", r"观察", r"наблюден", r"бақылау")),
+    ("document", (r"\barchival\b", r"\bdocument(?:s|ary)?\b", r"档案", r"文档", r"文件分析", r"архив", r"документ", r"мұрағат", r"құжат")),
+    ("field_note", (r"\bfield\s+notes?\b", r"田野笔记", r"полев.{0,6}замет", r"далалық.{0,6}жазб")),
+    ("digital_trace", (r"\bdigital\s+traces?\b", r"数字痕迹", r"цифров.{0,6}след", r"цифрлық.{0,6}із")),
+    ("log", (r"\blogs?\b", r"日志", r"журнал.{0,8}лог", r"лог", r"журнал")),
+)
+
+
+def _contains_any(text: str, patterns: tuple[str, ...]) -> bool:
+    return any(re.search(pattern, text, flags=re.I | re.UNICODE) for pattern in patterns)
+
+
+def build_user_research_intent(topic: str, keywords: Any, *, mode: str = "") -> dict[str, Any]:
+    """Freeze user topic/keywords as research-intent authority, separate from literature evidence."""
+    source_topic = re.sub(r"\s+", " ", str(topic or "")).strip()
+    source_keywords = _verbatim_keywords(keywords)
+    text = " ; ".join([source_topic] + source_keywords)
+    relation_types: list[str] = []
+    if _contains_any(text, _MODERATION_CUES):
+        relation_types.append("moderation")
+    if _contains_any(text, _MEDIATION_CUES):
+        relation_types.append("mediation")
+    if _contains_any(text, _SERIAL_MEDIATION_CUES):
+        relation_types.append("serial_mediation")
+    if _contains_any(text, _PARALLEL_MEDIATION_CUES):
+        relation_types.append("parallel_mediation")
+
+    qualitative_design = ""
+    for canonical, patterns in _QUAL_DESIGN_PATTERNS:
+        if _contains_any(text, patterns):
+            qualitative_design = canonical
+            break
+    source_families: list[str] = []
+    for family, patterns in _QUAL_SOURCE_PATTERNS:
+        if _contains_any(text, patterns):
+            source_families.append(family)
+
+    return {
+        "version": USER_RESEARCH_INTENT_VERSION,
+        "mode": str(mode or "").strip().casefold(),
+        "source_topic": source_topic,
+        "source_keywords": source_keywords,
+        "explicit_relation_types": relation_types,
+        "qualitative_design": qualitative_design,
+        "qualitative_data_source_families": source_families,
+        "preservation_policy": "user_topic_keywords_and_explicit_design_are_authoritative;literature_cannot_silently_redefine_scope",
+        "evidence_policy": "verified_sources_support_prior_literature;author_inference_may_develop_novel_logic_without_fake_citations",
+    }
+
+
+def _cue_positions(text: str, patterns: tuple[str, ...]) -> list[int]:
+    positions: list[int] = []
+    for pattern in patterns:
+        positions.extend(match.start() for match in re.finditer(pattern, text, flags=re.I | re.UNICODE))
+    return positions
+
+
+
+def _phrase_bound_intent_label(text: str, labels: list[str], relation: str) -> str:
+    """Resolve an explicitly named mediator/moderator from multilingual cue grammar."""
+    for label in labels:
+        clean = re.sub(r"\s+", " ", str(label or "")).strip()
+        if not clean:
+            continue
+        esc = re.escape(clean)
+        if relation == "moderation":
+            patterns = (
+                rf"(?i)\bmoderating\s+role\s+of\s+(?:the\s+)?{esc}\b",
+                rf"(?i)\bmoderation(?:\s+effect)?\s+of\s+(?:the\s+)?{esc}\b",
+                rf"(?i)\b{esc}\b.{0,24}\b(?:as\s+(?:a\s+)?)?moderator\b",
+                rf"{esc}(?:的)?调节(?:作用|效应|变量|关系)?",
+                rf"(?i)модерирующ\w*\s+роль\s+(?:\S+\s+){{0,3}}?{esc}",
+                rf"(?i){esc}.{{0,12}}модератор ретінде",
+            )
+        else:
+            patterns = (
+                rf"(?i)\bmediating\s+role\s+of\s+(?:the\s+)?{esc}\b",
+                rf"(?i)\bmediation(?:\s+effect)?\s+of\s+(?:the\s+)?{esc}\b",
+                rf"(?i)\b{esc}\b.{0,24}\b(?:as\s+(?:a\s+)?)?mediator\b",
+                rf"{esc}(?:的)?(?:链式|并行)?中介(?:作用|效应|变量)?",
+                rf"(?i)медиирующ\w*\s+роль\s+(?:\S+\s+){{0,3}}?{esc}",
+                rf"(?i){esc}.{{0,12}}медиатор ретінде",
+            )
+        if any(re.search(pattern, text, flags=re.UNICODE) for pattern in patterns):
+            return clean
+    return ""
+
+def _closest_intent_label(text: str, labels: list[str], cue_patterns: tuple[str, ...]) -> str:
+    """Choose the user-mentioned label nearest an explicit relation cue; never invent a label."""
+    folded = text.casefold()
+    cue_pos = _cue_positions(text, cue_patterns)
+    if not cue_pos:
+        return ""
+    scored: list[tuple[int, int, str]] = []
+    for order, label in enumerate(labels):
+        clean = re.sub(r"\s+", " ", str(label or "")).strip()
+        if not clean:
+            continue
+        start = folded.find(clean.casefold())
+        if start < 0:
+            continue
+        midpoint = start + max(1, len(clean) // 2)
+        distance = min(abs(midpoint - pos) for pos in cue_pos)
+        scored.append((distance, order, clean))
+    return min(scored)[2] if scored else ""
+
+
+def _ensure_construct_role(constructs: list[dict[str, Any]], label: str, role: str) -> str:
+    if not label:
+        return ""
+    for row in constructs:
+        if str(row.get("label") or "").casefold() == label.casefold():
+            roles = list(row.get("roles") or [])
+            if role not in roles:
+                roles = [r for r in roles if not (role in {"mediator", "moderator"} and r in {"control", "covariate"})]
+                roles.append(role)
+                row["roles"] = roles
+            return str(row.get("construct_id") or "")
+    cid = f"C{len(constructs) + 1:03d}"
+    constructs.append({
+        "construct_id": cid,
+        "label": label,
+        "roles": [role],
+        "measurement_model": "reflective",
+        "measurement_allowed": True,
+    })
+    return cid
+
+
+def _next_hypothesis_order(relations: list[dict[str, Any]]) -> int:
+    values = []
+    for row in relations:
+        try:
+            if row.get("hypothesis_order") is not None:
+                values.append(int(row.get("hypothesis_order")))
+        except Exception:
+            pass
+    numbered_count = sum(1 for row in relations if row.get("numbered_hypothesis", True) and row.get("type") != "control_effect")
+    return max(max(values) if values else 0, numbered_count) + 1
+
+
+def _append_relation(relations: list[dict[str, Any]], *, rtype: str, source: str, target: str, moderator: str = "") -> None:
+    if not source or not target or source == target:
+        return
+    for row in relations:
+        if row.get("type") != rtype or row.get("source") != source or row.get("target") != target:
+            continue
+        if rtype != "moderation" or row.get("moderator") == moderator:
+            return
+    relations.append({
+        "relation_id": f"R{len(relations) + 1:03d}",
+        "type": rtype,
+        "source": source,
+        "target": target,
+        "moderator": moderator or None,
+        "analysis_target": True,
+        "numbered_hypothesis": rtype != "control_effect",
+        "hypothesis_order": _next_hypothesis_order(relations),
+    })
+
+
+def _reconcile_explicit_quantitative_intent(
+    *, topic: str, keywords: Any, constructs: list[dict[str, Any]], relations: list[dict[str, Any]], decisions: list[dict[str, Any]]
+) -> dict[str, Any]:
+    """Deterministically close omissions that would silently erase explicit user design intent."""
+    intent = build_user_research_intent(topic, keywords, mode="quantitative")
+    explicit = set(intent.get("explicit_relation_types") or [])
+    if not explicit:
+        return intent
+
+    intent_text = " ; ".join([intent.get("source_topic") or ""] + list(intent.get("source_keywords") or []))
+    labels = [str(row.get("label") or "") for row in constructs]
+    labels += [kw for kw in (intent.get("source_keywords") or []) if kw.casefold() not in {x.casefold() for x in labels}]
+
+    def ids_for(role: str) -> list[str]:
+        return [str(row.get("construct_id") or "") for row in constructs if role in (row.get("roles") or [])]
+
+    if "moderation" in explicit:
+        moderator_ids = ids_for("moderator")
+        # Prefer the construct named nearest the user's explicit moderation cue.
+        # A model-assigned role is advisory; the user's title/keywords are authority.
+        moderator_label = _phrase_bound_intent_label(intent_text, labels, "moderation")
+        if not moderator_label:
+            noncore_labels = [
+                str(row.get("label") or "") for row in constructs
+                if not set(row.get("roles") or []).intersection({"predictor", "mediator", "outcome"})
+            ]
+            if len([label for label in noncore_labels if label]) == 1:
+                moderator_label = next(label for label in noncore_labels if label)
+        if not moderator_label:
+            moderator_label = _closest_intent_label(intent_text, labels, _MODERATION_CUES)
+        if moderator_label:
+            moderator_id = _ensure_construct_role(constructs, moderator_label, "moderator")
+        elif len(moderator_ids) == 1:
+            moderator_id = moderator_ids[0]
+        else:
+            moderator_id = ""
+        if moderator_id:
+            direct = [r for r in relations if r.get("type") == "direct_association"]
+            predictor_ids, outcome_ids = ids_for("predictor"), ids_for("outcome")
+            preferred = next((r for r in direct if r.get("source") in predictor_ids and r.get("target") in outcome_ids), None)
+            anchor = preferred or (direct[0] if direct else None)
+            source = str((anchor or {}).get("source") or (predictor_ids[0] if predictor_ids else ""))
+            target = str((anchor or {}).get("target") or (outcome_ids[0] if outcome_ids else ""))
+            before = len(relations)
+            _append_relation(relations, rtype="moderation", source=source, target=target, moderator=moderator_id)
+            if len(relations) > before:
+                decisions.append({"reason":"explicit_user_moderation_reconciled","moderator":moderator_id,"source":source,"target":target})
+
+    if "mediation" in explicit or "serial_mediation" in explicit or "parallel_mediation" in explicit:
+        mediator_ids = ids_for("mediator")
+        mediator_label = _phrase_bound_intent_label(intent_text, labels, "mediation") or _closest_intent_label(intent_text, labels, _MEDIATION_CUES)
+        if mediator_label:
+            mediator_id = _ensure_construct_role(constructs, mediator_label, "mediator")
+            mediator_ids = [mediator_id] + [mid for mid in mediator_ids if mid != mediator_id]
+        elif not mediator_ids:
+            mediator_ids = []
+        predictor_ids, outcome_ids = ids_for("predictor"), ids_for("outcome")
+        if predictor_ids and outcome_ids and mediator_ids:
+            p, o = predictor_ids[0], outcome_ids[0]
+            if "serial_mediation" in explicit and len(mediator_ids) > 1:
+                path = [p] + mediator_ids + [o]
+                for a, b in zip(path, path[1:]):
+                    before = len(relations); _append_relation(relations, rtype="direct_association", source=a, target=b)
+                    if len(relations) > before:
+                        decisions.append({"reason":"explicit_user_serial_mediation_edge_reconciled","source":a,"target":b})
+            else:
+                for mediator_id in mediator_ids:
+                    for a, b in ((p, mediator_id), (mediator_id, o)):
+                        before = len(relations); _append_relation(relations, rtype="direct_association", source=a, target=b)
+                        if len(relations) > before:
+                            decisions.append({"reason":"explicit_user_mediation_edge_reconciled","source":a,"target":b})
+    return intent
+
 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()
@@ -178,6 +462,16 @@ def build_quantitative_contract(
             relation["hypothesis_order"] = None
         relations.append(relation)
 
+    # R7-R2C-R5: user topic/keywords are research-intent authority.  Literature
+    # availability and an incomplete model-proposed JSON may not silently erase an
+    # explicitly requested mediation/moderation design.  Reconciliation is
+    # deterministic and never invents literature evidence.
+    user_intent = _reconcile_explicit_quantitative_intent(
+        topic=topic, keywords=keywords, constructs=constructs, relations=relations, decisions=rejected
+    )
+    by_id = {row["construct_id"]: row for row in constructs}
+    label_to_id = {str(row.get("label") or "").casefold(): str(row.get("construct_id") or "") for row in constructs}
+
     # 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:
@@ -254,8 +548,9 @@ def build_quantitative_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),
+        "source_topic": str(user_intent.get("source_topic") or re.sub(r"\s+", " ", str(topic or "")).strip()),
+        "source_keywords": list(user_intent.get("source_keywords") or _verbatim_keywords(keywords)),
+        "user_research_intent": user_intent,
         "research_problem": _clean(raw.get("research_problem"), 500),
         "research_questions": _keywords(raw.get("research_questions") or []),
         "design_family": "cross_sectional_observational_survey",
@@ -292,6 +587,7 @@ def validate_quantitative_contract(contract: dict[str, Any]) -> list[str]:
     ids = {x.get("construct_id") for x in constructs}
     if len(ids) != len(constructs) or len(constructs) < 2:
         issues.append("invalid_construct_identity")
+    relations = [r for r in (quant.get("relations") or []) if isinstance(r, dict)]
     for r in quant.get("relations") or []:
         if not isinstance(r, dict):
             issues.append("invalid_relation_record"); continue
@@ -299,6 +595,17 @@ def validate_quantitative_contract(contract: dict[str, Any]) -> list[str]:
             issues.append("relation_references_unknown_construct")
         if r.get("type") == "moderation" and r.get("moderator") not in ids:
             issues.append("moderation_references_unknown_moderator")
+
+    # Semantic completeness is part of contract validity: an internally legal graph
+    # is still invalid when it silently drops an explicit user-requested relation.
+    intent = dict(record.get("user_research_intent") or {})
+    explicit = set(intent.get("explicit_relation_types") or [])
+    if "moderation" in explicit and not any(r.get("type") == "moderation" for r in relations):
+        issues.append("explicit_user_moderation_missing")
+    if explicit.intersection({"mediation", "serial_mediation", "parallel_mediation"}) and not (quant.get("indirect_effects") or []):
+        issues.append("explicit_user_mediation_missing")
+    if str(record.get("source_topic") or "") != str(intent.get("source_topic") or record.get("source_topic") or ""):
+        issues.append("user_topic_authority_drift")
     return list(dict.fromkeys(issues))
 
 
@@ -374,6 +681,7 @@ def build_quantitative_front_matter_contract(contract: dict[str, Any], *, versio
 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 {})
+    user_intent = build_user_research_intent(topic, keywords, mode="qualitative")
     sources = []
     for index, raw in enumerate(facts.get("data_sources") or [], start=1):
         label = _clean(raw, 120)
@@ -405,14 +713,20 @@ def build_qualitative_contract(topic: str, keywords: Any, metadata: dict[str, An
             "has_deviant_evidence": bool(item.get("has_deviant_evidence")),
         })
     design = _clean(facts.get("analytic_design") or facts.get("analysis_method"), 120)
+    # For system-generated qualitative evidence, explicit user method intent owns
+    # the design label.  Verified user-provided material remains empirical truth
+    # and is never rewritten to satisfy a title/keyword request.
+    if str(meta.get("source_type") or "").strip().casefold() != "user_provided":
+        design = _clean(user_intent.get("qualitative_design") or design, 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),
+        "source_topic": str(user_intent.get("source_topic") or re.sub(r"\s+", " ", str(topic or "")).strip()),
+        "source_keywords": list(user_intent.get("source_keywords") or _verbatim_keywords(keywords)),
+        "user_research_intent": user_intent,
         "research_problem": "",
         "research_questions": [],
         "design_family": design or "Iterative thematic/process analysis",