A5-R3-R7-R2C-R6B policy-driven keyword authority scoped deployment

core/generation_runner.py

@@ -32,7 +32,12 @@ 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 core.study_design_contract import (
+    build_user_research_intent,
+    keyword_policy_from_contract,
+    apply_keyword_policy,
+    parse_keyword_terms,
+)
 from utils.file_parser import extract_text_from_files
 
 ProgressCallback = Callable[..., None]
@@ -1796,27 +1801,73 @@ class GenerationRunner:
         return str(topic or "").strip(), str(keywords or "").strip()
 
     @staticmethod
-    def _canonical_quantitative_front_matter_keywords(mode: str, candidate_keywords, data_metadata: dict):
-        """Return user-authoritative keywords when available; otherwise normalize a legacy candidate.
+    def _resolved_front_matter_keyword_policy(data_metadata: dict):
+        """Resolve one policy object for Working/Final/retry/resume/fallback paths."""
+        meta = dict(data_metadata or {})
+        contract = dict(meta.get("study_design_contract") or meta.get("front_matter_contract") or {})
+        if not contract and isinstance(meta.get("user_research_intent"), dict):
+            contract = {"user_research_intent": dict(meta.get("user_research_intent") or {})}
+        return keyword_policy_from_contract(contract)
 
-        The name is retained for checkpoint compatibility, but R7-R2C-R5 applies
-        the same authority rule to quantitative and qualitative manuscripts.
-        """
+    @staticmethod
+    def _front_matter_keyword_fallback_terms(data_metadata: dict):
+        """Derive recoverable generated keywords only from frozen study-design authority."""
         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 source:
-            value = re.sub(r"\s+", " ", str(item or "")).strip()
-            key = value.casefold()
-            if value and key not in seen:
-                values.append(value)
-                seen.add(key)
+        quant = dict(contract.get("quantitative") or {})
+        for row in quant.get("constructs") or []:
+            if isinstance(row, dict):
+                label = re.sub(r"\s+", " ", str(row.get("label") or "")).strip()
+                if label:
+                    values.append(label)
+        qual = dict(contract.get("qualitative") or {})
+        if qual:
+            for candidate in (
+                qual.get("phenomenon"),
+                qual.get("design_tradition"),
+            ):
+                value = re.sub(r"\s+", " ", str(candidate or "")).strip()
+                if value:
+                    values.append(value)
+            for theme in meta.get("themes") or []:
+                if isinstance(theme, dict):
+                    value = re.sub(r"\s+", " ", str(theme.get("title") or "")).strip()
+                    if value:
+                        values.append(value)
+        return parse_keyword_terms(values, explicit_user=False)
+
+    @staticmethod
+    def _user_authoritative_front_matter_keywords(data_metadata: dict):
+        """Return explicit user keywords only when the resolved policy says USER_EXPLICIT."""
+        meta = dict(data_metadata or {})
+        contract = dict(meta.get("study_design_contract") or meta.get("front_matter_contract") or {})
+        intent = dict(contract.get("user_research_intent") or meta.get("user_research_intent") or {})
+        policy = keyword_policy_from_contract(contract)
+        if str(policy.get("authority_source") or "").upper() != "USER_EXPLICIT":
+            return []
+        return apply_keyword_policy(
+            intent.get("source_keywords") or [],
+            policy,
+            authoritative=intent.get("source_keywords") or [],
+        )
+
+    @staticmethod
+    def _canonical_quantitative_front_matter_keywords(mode: str, candidate_keywords, data_metadata: dict):
+        """Apply one keyword policy across quantitative/qualitative and all product tiers."""
+        meta = dict(data_metadata or {})
+        contract = dict(meta.get("study_design_contract") or meta.get("front_matter_contract") or {})
+        intent = dict(contract.get("user_research_intent") or meta.get("user_research_intent") or {})
+        policy = keyword_policy_from_contract(contract)
+        authoritative = list(intent.get("source_keywords") or [])
+        values = apply_keyword_policy(
+            candidate_keywords,
+            policy,
+            authoritative=authoritative if str(policy.get("authority_source") or "").upper() == "USER_EXPLICIT" else None,
+        )
         return values
 
+
     @staticmethod
     def _replace_title(manuscript: str, title: str) -> str:
         value = str(manuscript or "")
@@ -2014,7 +2065,7 @@ class GenerationRunner:
         # 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()]
+            intent_keywords = parse_keyword_terms(keywords, explicit_user=True)
         for seed in intent_keywords:
             if 1 <= len(str(seed).split()) <= 10:
                 add(seed, 'user_intent_keyword', 'STRICT_CONSTRUCT')
@@ -3493,6 +3544,9 @@ class GenerationRunner:
                 topic, keywords, research_logic_contract, synthesis_result,
                 data_evidence=planning_data_md, mode=mode,
                 hypothesis_contract=hypothesis_contract,
+                authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
+                keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
+                fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
             )
             working_abstract = str(working_front.get("abstract") or "").strip()
             if hypothesis_ledger:
@@ -3522,6 +3576,9 @@ class GenerationRunner:
                     topic, keywords, research_logic_contract, synthesis_result,
                     data_evidence=planning_data_md, mode=mode, validation_feedback=feedback,
                     hypothesis_contract=hypothesis_contract,
+                    authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
+                    keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
+                    fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
                 )
                 working_abstract = str(working_front.get("abstract") or "").strip()
                 if hypothesis_ledger:
@@ -3852,6 +3909,9 @@ class GenerationRunner:
                     topic, keywords, research_logic_contract, synthesis_result,
                     data_evidence=planning_data_md, mode=mode, validation_feedback=feedback,
                     hypothesis_contract=hypothesis_contract,
+                    authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
+                    keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
+                    fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
                 )
                 candidate_abstract = str(working_front.get("abstract") or "").strip()
                 if hypothesis_ledger:
@@ -4078,6 +4138,9 @@ class GenerationRunner:
                 topic, keywords, research_logic_contract, synthesis_result,
                 data_evidence=planning_data_md, mode=mode,
                 hypothesis_contract=hypothesis_contract,
+                authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
+                keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
+                fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
             )
             working_abstract = str(working_front.get("abstract") or "").strip()
             if hypothesis_ledger:
@@ -4102,6 +4165,9 @@ class GenerationRunner:
                     topic, keywords, research_logic_contract, synthesis_result,
                     data_evidence=planning_data_md, mode=mode, validation_feedback=feedback,
                     hypothesis_contract=hypothesis_contract,
+                    authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
+                    keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
+                    fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
                 )
                 working_abstract = str(working_front.get("abstract") or "").strip()
                 if hypothesis_ledger:
@@ -5243,6 +5309,9 @@ The previous attempt for this same manuscript unit was rejected by validation. R
             final_front = agent.generate_final_front_matter(
                 topic, keywords, body_only, data_evidence=user_data_md, mode=mode,
                 hypothesis_contract=hypothesis_contract,
+                authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
+                keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
+                fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
             )
             final_abstract = str(final_front.get("abstract") or "").strip()
             if hypothesis_ledger:
@@ -5272,6 +5341,9 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                 final_front = agent.generate_final_front_matter(
                     topic, keywords, body_only, data_evidence=user_data_md, mode=mode,
                     validation_feedback=feedback, hypothesis_contract=hypothesis_contract,
+                    authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
+                    keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
+                    fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
                 )
                 final_abstract = str(final_front.get("abstract") or "").strip()
                 if hypothesis_ledger:
@@ -5319,12 +5391,13 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                 )
                 if candidate_title:
                     final_title = candidate_title
-                if 4 <= len(candidate_keywords) <= 6:
+                if candidate_keywords:
+                    # R7-R2C-R6B: the resolved policy owns keyword identity/count
+                    # semantics. No local numeric cardinality branch is permitted.
                     final_keywords = candidate_keywords
                 else:
                     logging.warning(
-                        "Adaptive final keyword reconciliation returned nonpreferred count=%s; preserving validated Final Front Matter keywords.",
-                        len(candidate_keywords),
+                        "Adaptive final keyword reconciliation returned no usable keywords; preserving validated Final Front Matter keywords."
                     )
             full_draft_en = replace_abstract(full_draft_en, final_abstract)
             full_draft_en = self._replace_title(full_draft_en, final_title)

core/pipeline_agent.py

@@ -28,7 +28,14 @@ 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.study_design_contract import (
+    inspect_study_design_scope,
+    resolve_keyword_policy,
+    keyword_policy_from_contract,
+    apply_keyword_policy,
+    evaluate_keyword_count,
+    parse_keyword_terms,
+)
 from core.model_gateway import ProviderNeutralModelGateway
 from core.model_usage import ModelUsageEvent, emit_usage_event, normalize_deepseek_usage, prompt_identity, sanitize_metadata
 
@@ -1612,22 +1619,27 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
         return _stream_direct_research_logic()
 
     @staticmethod
-    def _front_matter_keywords(value, minimum=4, maximum=6):
-        if isinstance(value, list):
-            raw = value
-        else:
-            raw = re.split(r'[,;\n]+', str(value or ''))
-        cleaned = []
-        for item in raw:
-            term = re.sub(r'^[\s\-•]+|[\s.;]+$', '', str(item or '')).strip()
-            if term and term.casefold() not in {existing.casefold() for existing in cleaned}:
-                cleaned.append(term)
-        if not minimum <= len(cleaned) <= maximum:
-            raise DeepSeekRequestError(
-                f'Front-matter keywords must contain {minimum}-{maximum} distinct terms; got {len(cleaned)}.'
+    def _front_matter_keywords(value, authoritative=None, keyword_policy=None, fallback=None):
+        """Normalize keywords through one resolved policy; count is never a generation kill switch."""
+        policy = dict(keyword_policy or resolve_keyword_policy(authoritative))
+        cleaned = apply_keyword_policy(
+            value,
+            policy,
+            authoritative=authoritative,
+            fallback=fallback,
+        )
+        if not cleaned:
+            raise DeepSeekRequestError('Front-matter keywords are empty and no deterministic fallback is available.')
+        status = evaluate_keyword_count(cleaned, policy)
+        if status.get('count_mode') != 'UNSPECIFIED' and not status.get('within_policy_range'):
+            logging.warning(
+                'Front-matter keyword count advisory | count=%s mode=%s minimum=%s maximum=%s failure_mode=%s nonfatal=true',
+                status.get('count'), status.get('count_mode'), status.get('minimum'), status.get('maximum'),
+                status.get('failure_mode'),
             )
         return cleaned
 
+
     @staticmethod
     def _front_matter_has_citation(value):
         text = str(value or '')
@@ -1696,11 +1708,39 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
         phase,
         max_tokens,
         validator,
+        authoritative_keywords=None,
+        keyword_policy=None,
     ):
         """Generate one front-matter object with one bounded content-only retry."""
         base_prompt = str(user_prompt)
         current_prompt = base_prompt
         last_error = None
+        policy = dict(keyword_policy or resolve_keyword_policy(authoritative_keywords))
+        authority_terms = apply_keyword_policy(
+            authoritative_keywords,
+            policy,
+            authoritative=authoritative_keywords,
+        ) if authoritative_keywords else []
+        if authority_terms:
+            keyword_retry_rule = (
+                'a JSON array field "keywords" that preserves every user-supplied keyword identity and order. '
+                'Do not add, delete, merge, truncate, resize, or reinterpret the authoritative list.'
+            )
+        else:
+            count_mode = str(policy.get('count_mode') or 'UNSPECIFIED').upper()
+            minimum = policy.get('preferred_min') if count_mode == 'PREFERRED' else policy.get('required_min')
+            maximum = policy.get('preferred_max') if count_mode == 'PREFERRED' else policy.get('required_max')
+            if count_mode in {'PREFERRED', 'REQUIRED'} and (minimum is not None or maximum is not None):
+                keyword_retry_rule = (
+                    'a JSON array field "keywords" containing distinct non-empty academic terms aligned with the frozen study design. '
+                    f'The resolved keyword policy is {count_mode} with minimum={minimum} and maximum={maximum}; '
+                    'treat count as advisory/export-compliance metadata rather than a manuscript-generation failure.'
+                )
+            else:
+                keyword_retry_rule = (
+                    'a JSON array field "keywords" containing a concise, distinct set of non-empty academic terms '
+                    'aligned with the frozen study design. No numeric keyword count is implied by the journal-style label.'
+                )
         for local_attempt in range(2):
             # API/auth/balance/cancellation/transport behavior remains owned by
             # call_deepseek(). Those failures are deliberately not caught here.
@@ -1750,14 +1790,14 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
                         'supported/rejected hypotheses, statistical significance, coefficients, p-values, fit '
                         'statistics, substantive findings, or conclusions. Use prospective/non-result language such '
                         'as examines, investigates, asks, develops, evaluates, or is designed to assess. Keep exactly '
-                        'a non-empty string field "abstract" and a JSON array field "keywords" containing 4-6 distinct '
-                        'non-empty academic terms. Do not describe the correction and do not wrap the JSON in Markdown.'
+                        'a non-empty string field "abstract" and ' + keyword_retry_rule + ' '
+                        'Do not describe the correction and do not wrap the JSON in Markdown.'
                     )
                 else:
                     current_prompt = base_prompt + (
                         '\n\nLOCAL VALIDATION RETRY: The previous response failed local JSON/schema/content validation. '
                         'Return a fresh complete JSON object only, with exactly a non-empty string field "abstract" '
-                        'and a JSON array field "keywords" containing 4-6 distinct non-empty academic terms. '
+                        'and ' + keyword_retry_rule + ' '
                         'Do not describe the correction and do not wrap the JSON in Markdown.'
                     )
         raise DeepSeekRequestError(f'{phase} front matter validation failed: {last_error}')
@@ -1788,28 +1828,28 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
 
     @staticmethod
     def _quantitative_front_matter_fallback(contract):
-        """Last-resort degraded front matter derived only from frozen authority.
-
-        This fallback is intentionally plain.  It is used only after bounded
-        semantic reconciliation cannot produce a validated result; it never
-        replaces the normal LLM path.
-        """
-        items = SSciPipelineAgent._quantitative_front_matter_items(contract)
+        """Deterministic keyword fallback from frozen study-design authority, with no count target."""
+        record = dict(contract or {})
+        policy = keyword_policy_from_contract(record)
+        intent = dict(record.get('user_research_intent') or {})
+        authority = list(intent.get('source_keywords') or [])
+        items = SSciPipelineAgent._quantitative_front_matter_items(record)
         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 "")
-        keywords = list(values)
-        for item in ("quantitative research", "empirical study", "management research"):
-            if len(keywords) >= 4:
-                break
-            if item.casefold() not in {value.casefold() for value in keywords}:
-                keywords.append(item)
-        return {"title": title, "keywords": keywords[:6]}
+        if authority:
+            keywords = apply_keyword_policy(authority, policy, authoritative=authority)
+        else:
+            keywords = apply_keyword_policy(values, policy)
+        if not keywords:
+            keywords = ["quantitative management research"]
+        if values:
+            title = ", ".join(values[:-1]) + ((", and " if len(values) > 2 else " and ") + values[-1] if len(values) > 1 else "")
+        else:
+            title = str(intent.get('source_topic') or record.get('source_topic') or 'Quantitative Management Study').strip()
+        return {"title": title, "keywords": keywords}
 
     @staticmethod
     def _validate_quantitative_front_matter_candidate(candidate, contract):
-        """Validate structure and frozen-construct coverage, not prose style."""
+        """Validate structure and frozen-construct coverage; keyword count is policy metadata, not a fatal schema."""
         if not isinstance(candidate, dict):
             return ["candidate_not_object"]
         title = re.sub(r"\s+", " ", str(candidate.get("title") or "")).strip()
@@ -1818,13 +1858,10 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
             return ["title_missing"]
         if not isinstance(raw_keywords, list):
             return ["keywords_not_array"]
-        keywords = [re.sub(r"\s+", " ", str(item or "")).strip() for item in raw_keywords]
-        keywords = [item for item in keywords if item]
+        keywords = parse_keyword_terms(raw_keywords, explicit_user=False)
         issues = []
-        if not 4 <= len(keywords) <= 6:
-            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")
+        if not keywords:
+            issues.append("keywords_empty")
         frozen_items = SSciPipelineAgent._quantitative_front_matter_items(contract)
         combined = (title + " ; " + " ; ".join(keywords)).casefold()
         missing = [
@@ -1837,6 +1874,7 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
             issues.append("missing_frozen_constructs:" + " | ".join(missing))
         return issues
 
+
     def reconcile_quantitative_reader_front_matter(self, *, topic, keywords, contract, manuscript=""):
         """Preserve R5 user intent while retaining legacy contract reconciliation.
 
@@ -1848,17 +1886,17 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
         """
         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()]
+        policy = keyword_policy_from_contract(record, fallback_keywords=keywords)
+        source_topic = str(intent.get("source_topic") or record.get("source_topic") or topic or "").strip()
+        source_keywords = list(intent.get("source_keywords") or [])
+        if str(policy.get("authority_source") or "").upper() == "USER_EXPLICIT" and source_keywords:
             return {
                 "title": source_topic,
-                "keywords": [re.sub(r"\s+", " ", str(item or "")).strip() for item in source_keywords if str(item or "").strip()],
+                "keywords": apply_keyword_policy(source_keywords, policy, authoritative=source_keywords),
             }
 
-        # Legacy compatibility: no R5 user-intent authority was frozen.
+        # No explicit user keyword authority: retain user-authoritative title when
+        # available, while resolving generated keywords from frozen empirical scope.
         """Adaptively reconcile final AMJ title/keywords to frozen evidence.
 
         Original user topic/keyword seeds are retained as research context.  The
@@ -1867,16 +1905,17 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
         validates semantic fidelity.  After bounded retries, deterministic
         degraded-mode fallback keeps packaging recoverable.
         """
-        contract = dict(contract or {})
-        frozen = self._quantitative_front_matter_items(contract)
+        frozen = self._quantitative_front_matter_items(record)
         if not frozen:
             logging.warning("Front-matter semantic contract missing frozen constructs; using degraded fallback.")
-            return self._quantitative_front_matter_fallback(contract)
+            fallback = self._quantitative_front_matter_fallback(record)
+            if intent and source_topic:
+                fallback["title"] = source_topic
+            return fallback
 
-        source_topic = str(contract.get("source_topic") or topic or "").strip()
-        source_keywords = list(contract.get("source_keywords") or [])
+        source_keywords = list(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()]
+            source_keywords = parse_keyword_terms(keywords, explicit_user=True)
         body_context = self._abstract_manuscript_context(str(manuscript or ""), 8000) if manuscript else ""
         feedback = ""
         last_candidate = None
@@ -1891,7 +1930,7 @@ SEMANTIC RULES:
 3. Do NOT present any concept outside the frozen construct-role list as a focal measured, modeled, mediating, moderating, or outcome variable.
 4. Contextual terms may appear when they describe setting/domain rather than an unmeasured focal construct.
 5. Produce a natural journal-style title, not a mechanical template.
-6. Return 4-6 distinct academic keywords. Include the frozen constructs and add only genuinely useful context/method terms.
+6. Return a concise distinct keyword set aligned with the frozen constructs and genuinely useful context/method terms. Do not infer a numeric keyword count from the journal-style label; count guidance comes only from the resolved keyword policy.
 7. Do not mention this reconciliation process.
 
 <ORIGINAL_USER_TOPIC>
@@ -1960,9 +1999,10 @@ A contextual/domain term is allowed. Mark INVALID only when the proposal makes a
             extra = [str(item).strip() for item in ((critic or {}).get("unmeasured_focal_constructs") or []) if str(item).strip()]
             missing = [str(item).strip() for item in ((critic or {}).get("missing_frozen_constructs") or []) if str(item).strip()]
             if not invalid and not extra and not missing:
+                resolved_title = source_topic if intent else re.sub(r"\s+", " ", str(candidate.get("title") or "")).strip()
                 return {
-                    "title": re.sub(r"\s+", " ", str(candidate.get("title") or "")).strip(),
-                    "keywords": [re.sub(r"\s+", " ", str(item)).strip() for item in candidate.get("keywords") or [] if str(item).strip()],
+                    "title": resolved_title,
+                    "keywords": apply_keyword_policy(candidate.get("keywords") or [], policy),
                 }
             feedback = "Semantic critic rejected candidate. " + str((critic or {}).get("reason") or "")
             if extra:
@@ -1971,13 +2011,15 @@ A contextual/domain term is allowed. Mark INVALID only when the proposal makes a
                 feedback += " Missing frozen constructs: " + ", ".join(missing) + "."
             logging.warning("Adaptive front-matter semantic retry | attempt=%s/2 | %s", attempt + 1, feedback[:900])
 
-        fallback = self._quantitative_front_matter_fallback(contract)
+        fallback = self._quantitative_front_matter_fallback(record)
+        if intent and source_topic:
+            fallback["title"] = source_topic
         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=''
+        self, topic, keywords, research_logic_contract, synthesis_result, data_evidence='', mode='amj', validation_feedback='', hypothesis_contract='', authoritative_keywords=None, keyword_policy=None, fallback_keywords=None
     ):
         """Create a provisional visible abstract/keyword anchor without prewriting results."""
         evidence = self._truncate_context(str(data_evidence or ''), 3500)
@@ -2007,15 +2049,16 @@ If the immutable hypothesis contract is non-empty, preserve its H-identifiers an
 </LOCAL_VALIDATION_FEEDBACK>"""
         def validate_working(parsed):
             abstract = self._validate_working_abstract(parsed.get('abstract'))
-            working_keywords = self._front_matter_keywords(parsed.get('keywords'))
+            working_keywords = self._front_matter_keywords(parsed.get('keywords'), authoritative=authoritative_keywords, keyword_policy=keyword_policy, fallback=fallback_keywords)
             return {'abstract': abstract, 'keywords': working_keywords}
 
         return self._generate_structured_front_matter(
             'You are a senior management-journal editor creating provisional front matter without anticipating results.',
             prompt, phase='Working', max_tokens=1100, validator=validate_working,
+            authoritative_keywords=authoritative_keywords, keyword_policy=keyword_policy,
         )
 
-    def generate_final_front_matter(self, topic, keywords, manuscript, data_evidence='', mode='amj', validation_feedback='', hypothesis_contract=''):
+    def generate_final_front_matter(self, topic, keywords, manuscript, data_evidence='', mode='amj', validation_feedback='', hypothesis_contract='', authoritative_keywords=None, keyword_policy=None, fallback_keywords=None):
         """Generate final Abstract + Keywords only from the frozen completed body/evidence."""
         evidence = self._truncate_context(str(data_evidence or ''), 5000)
         body = self._abstract_manuscript_context(manuscript, 18000)
@@ -2045,12 +2088,13 @@ If the immutable hypothesis contract is non-empty, preserve its H-identifiers an
             meta_findings = self.inspect_authorial_ai_meta(abstract)
             if meta_findings:
                 raise DeepSeekRequestError('Final Abstract contains high-confidence authorial AI-writing meta-commentary.')
-            final_keywords = self._front_matter_keywords(parsed.get('keywords'))
+            final_keywords = self._front_matter_keywords(parsed.get('keywords'), authoritative=authoritative_keywords, keyword_policy=keyword_policy, fallback=fallback_keywords)
             return {'abstract': abstract, 'keywords': final_keywords}
 
         structured = self._generate_structured_front_matter(
             'You are a senior journal editor finalizing front matter from frozen manuscript evidence.',
             prompt, phase='Final', max_tokens=1200, validator=validate_final,
+            authoritative_keywords=authoritative_keywords, keyword_policy=keyword_policy,
         )
         abstract = structured['abstract']
         if self._front_matter_has_citation(abstract):

core/study_design_contract.py

@@ -27,11 +27,171 @@ def _clean(value: Any, maximum: int = 240) -> str:
     return text[:maximum]
 
 
-def _keywords(value: Any) -> list[str]:
-    raw = value if isinstance(value, (list, tuple)) else re.split(r"[,;\n|]+", str(value or ""))
+KEYWORD_POLICY_VERSION = "smartpaper-keyword-policy-v1"
+KEYWORD_SEPARATOR_POLICY = "SEMICOLON_NEWLINE_PIPE_PRIMARY_COMMA_FALLBACK"
+_KEYWORD_COUNT_MODES = {"UNSPECIFIED", "PREFERRED", "REQUIRED"}
+_KEYWORD_AUTHORITY_SOURCES = {"USER_EXPLICIT", "SYSTEM_GENERATED", "LEGACY"}
+
+
+def parse_keyword_terms(value: Any, *, explicit_user: bool = False, diagnostics: list[dict[str, Any]] | None = None) -> list[str]:
+    """Parse keyword identities without imposing an academic count limit.
+
+    Explicit user text uses semicolon/newline/pipe as primary separators so an
+    internal comma can remain part of a deliberate keyword phrase.  Comma is
+    used only as a compatibility fallback when no primary separator is present.
+    List/tuple input is already structurally separated and is never split again.
+    """
+    if isinstance(value, (list, tuple)):
+        raw = list(value)
+    else:
+        text = str(value or "")
+        if explicit_user and re.search(r"[;\n|]", text):
+            raw = re.split(r"[;\n|]+", text)
+        else:
+            raw = re.split(r"[,;\n|]+", text)
     out: list[str] = []
     seen: set[str] = set()
     for item in raw:
+        term = re.sub(r"\s+", " ", str(item or "")).strip()
+        key = term.casefold()
+        if not term:
+            continue
+        if key in seen:
+            if diagnostics is not None:
+                diagnostics.append({"type": "keyword_duplicate_deduped", "keyword": term})
+            continue
+        out.append(term)
+        seen.add(key)
+    return out
+
+
+def resolve_keyword_policy(
+    user_keywords: Any = None,
+    *,
+    authority_source: str = "",
+    configured_policy: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Resolve one versioned keyword policy without journal-count hardcoding.
+
+    Numeric preferences/requirements are accepted only when supplied by an
+    external/versioned policy record.  AMJ/ASQ style labels do not imply a count.
+    Generation never fails merely because a count is outside a preference.
+    """
+    user_terms = parse_keyword_terms(user_keywords, explicit_user=True)
+    source = str(authority_source or "").strip().upper()
+    if source not in _KEYWORD_AUTHORITY_SOURCES:
+        source = "USER_EXPLICIT" if user_terms else "SYSTEM_GENERATED"
+
+    configured = dict(configured_policy or {})
+    count_mode = str(configured.get("count_mode") or "UNSPECIFIED").strip().upper()
+    if count_mode not in _KEYWORD_COUNT_MODES:
+        count_mode = "UNSPECIFIED"
+
+    def _optional_nonnegative_int(name: str):
+        value = configured.get(name)
+        if value in (None, ""):
+            return None
+        try:
+            value = int(value)
+        except (TypeError, ValueError):
+            return None
+        return value if value >= 0 else None
+
+    preferred_min = _optional_nonnegative_int("preferred_min")
+    preferred_max = _optional_nonnegative_int("preferred_max")
+    required_min = _optional_nonnegative_int("required_min")
+    required_max = _optional_nonnegative_int("required_max")
+    if count_mode == "UNSPECIFIED":
+        preferred_min = preferred_max = required_min = required_max = None
+    elif count_mode == "PREFERRED":
+        required_min = required_max = None
+    elif count_mode == "REQUIRED":
+        preferred_min = preferred_max = None
+
+    return {
+        "version": KEYWORD_POLICY_VERSION,
+        "authority_source": source,
+        "identity_policy": "PRESERVE_USER_IDENTITIES" if source == "USER_EXPLICIT" else "NORMALIZE_DISTINCT_IDENTITIES",
+        "separator_policy": KEYWORD_SEPARATOR_POLICY,
+        "duplicate_policy": "DEDUPE_EXACT_WITH_DIAGNOSTIC",
+        "journal_requirement_source": str(configured.get("journal_requirement_source") or "").strip(),
+        "count_mode": count_mode,
+        "preferred_min": preferred_min,
+        "preferred_max": preferred_max,
+        "required_min": required_min,
+        "required_max": required_max,
+        "count_failure_mode": "EXPORT_COMPLIANCE" if count_mode == "REQUIRED" else "ADVISORY",
+        "fallback_source": "FROZEN_STUDY_DESIGN_CONTRACT",
+        "target_language_policy": "TRANSLATE_IDENTITY_SET_WITHOUT_RESIZE",
+    }
+
+
+def keyword_policy_from_contract(contract: dict[str, Any] | None, *, fallback_keywords: Any = None) -> dict[str, Any]:
+    record = dict(contract or {})
+    intent = dict(record.get("user_research_intent") or {})
+    user_terms = list(intent.get("source_keywords") or [])
+    configured = dict(record.get("keyword_policy") or intent.get("keyword_policy") or {})
+    authority_source = str(configured.get("authority_source") or "").strip()
+    if not user_terms and not authority_source:
+        # Legacy serialized contracts may have source_keywords but no explicit
+        # R5/R6 user-authority marker; do not silently promote them to user input.
+        return resolve_keyword_policy(None, authority_source="LEGACY", configured_policy=configured)
+    return resolve_keyword_policy(
+        user_terms if user_terms else fallback_keywords,
+        authority_source=authority_source,
+        configured_policy=configured,
+    )
+
+
+def evaluate_keyword_count(keywords: Any, policy: dict[str, Any] | None) -> dict[str, Any]:
+    """Return advisory/compliance status; never decides manuscript completion."""
+    terms = parse_keyword_terms(keywords, explicit_user=False)
+    resolved = dict(policy or resolve_keyword_policy())
+    mode = str(resolved.get("count_mode") or "UNSPECIFIED").upper()
+    count = len(terms)
+    minimum = resolved.get("preferred_min") if mode == "PREFERRED" else resolved.get("required_min")
+    maximum = resolved.get("preferred_max") if mode == "PREFERRED" else resolved.get("required_max")
+    within = True
+    if minimum is not None and count < int(minimum):
+        within = False
+    if maximum is not None and count > int(maximum):
+        within = False
+    return {
+        "count": count,
+        "count_mode": mode,
+        "within_policy_range": within,
+        "minimum": minimum,
+        "maximum": maximum,
+        "blocking_generation": False,
+        "failure_mode": str(resolved.get("count_failure_mode") or "ADVISORY"),
+    }
+
+
+def apply_keyword_policy(
+    value: Any,
+    policy: dict[str, Any] | None,
+    *,
+    authoritative: Any = None,
+    fallback: Any = None,
+) -> list[str]:
+    resolved = dict(policy or resolve_keyword_policy(authoritative))
+    source = str(resolved.get("authority_source") or "SYSTEM_GENERATED").upper()
+    if source == "USER_EXPLICIT":
+        terms = parse_keyword_terms(authoritative, explicit_user=True)
+        if terms:
+            return terms
+    terms = parse_keyword_terms(value, explicit_user=False)
+    if not terms and fallback is not None:
+        terms = parse_keyword_terms(fallback, explicit_user=False)
+    return terms
+
+
+def _keywords(value: Any) -> list[str]:
+    # Internal non-user lists retain bounded metadata normalization. This is not
+    # a reader-facing front-matter academic cardinality rule.
+    out: list[str] = []
+    seen: set[str] = set()
+    for item in parse_keyword_terms(value, explicit_user=False):
         text = _clean(item, 100)
         key = text.casefold()
         if text and key not in seen:
@@ -40,22 +200,12 @@ 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]
+    """Preserve explicit user keyword identity/order under the resolved separator policy."""
+    return parse_keyword_terms(value, explicit_user=True)
 
 
 _MODERATION_CUES = (
@@ -105,7 +255,9 @@ def _contains_any(text: str, patterns: tuple[str, ...]) -> bool:
 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)
+    keyword_diagnostics: list[dict[str, Any]] = []
+    source_keywords = parse_keyword_terms(keywords, explicit_user=True, diagnostics=keyword_diagnostics)
+    keyword_policy = resolve_keyword_policy(source_keywords)
     text = " ; ".join([source_topic] + source_keywords)
     relation_types: list[str] = []
     if _contains_any(text, _MODERATION_CUES):
@@ -132,6 +284,8 @@ def build_user_research_intent(topic: str, keywords: Any, *, mode: str = "") ->
         "mode": str(mode or "").strip().casefold(),
         "source_topic": source_topic,
         "source_keywords": source_keywords,
+        "keyword_policy": keyword_policy,
+        "keyword_diagnostics": keyword_diagnostics,
         "explicit_relation_types": relation_types,
         "qualitative_design": qualitative_design,
         "qualitative_data_source_families": source_families,
@@ -551,6 +705,7 @@ def build_quantitative_contract(
         "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,
+        "keyword_policy": dict(user_intent.get("keyword_policy") or resolve_keyword_policy()),
         "research_problem": _clean(raw.get("research_problem"), 500),
         "research_questions": _keywords(raw.get("research_questions") or []),
         "design_family": "cross_sectional_observational_survey",
@@ -727,6 +882,7 @@ def build_qualitative_contract(topic: str, keywords: Any, metadata: dict[str, An
         "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,
+        "keyword_policy": dict(user_intent.get("keyword_policy") or resolve_keyword_policy()),
         "research_problem": "",
         "research_questions": [],
         "design_family": design or "Iterative thematic/process analysis",