L2-R1 Unicode keyword authority and upstream quantitative contract repair

core/generation_runner.py

@@ -4374,9 +4374,25 @@ class GenerationRunner:
                 agent.global_memory["core_golden_metadata"] = extract_golden_metadata(agent, all_paper_texts)
             if mode == "amj":
                 data_file = data_paths[0] if data_paths else None
+                # L2-R1: establish the validated English identity projection
+                # BEFORE a system-generated quantitative construct/statistical
+                # contract is frozen. The immutable source Title/Keywords remain
+                # semantic authority and are passed separately below.
+                upstream_projection = self._resolve_english_front_matter_projection(
+                    agent, state, data_metadata, topic, keywords
+                )
+                internal_topic_en = str(
+                    upstream_projection.get("english_title") or topic or ""
+                ).strip()
+                internal_keywords_en = list(upstream_projection.get("english_keywords") or [])
                 _, user_data_md, _, data_metadata = quant_engine.resolve_empirical_data(
-                    data_file, interface_language, topic=topic, keywords=keywords, agent=agent
+                    data_file, interface_language,
+                    topic=topic, keywords=keywords,
+                    internal_topic=internal_topic_en,
+                    internal_keywords=internal_keywords_en,
+                    agent=agent,
                 )
+                data_metadata["front_matter_language_projection"] = dict(upstream_projection)
             else:
                 raw_interviews = extract_text_from_files(interview_paths[:1]) if interview_paths else ""
                 prewrite_started = time.monotonic()

core/quantitative_engine.py

@@ -474,7 +474,9 @@ class QuantitativeEngine:
             if isinstance(item, dict) and str(item.get("label") or "").strip()
         ]
 
-    def _derive_system_design(self, agent, topic="", keywords=""):
+    def _derive_system_design(
+        self, agent, topic="", keywords="", *, source_topic=None, source_keywords=None
+    ):
         """Interpret a parsimonious topic-aligned quantitative graph adaptively.
 
         The existing ``structured_json`` logical node is intentionally reused so
@@ -497,7 +499,10 @@ class QuantitativeEngine:
             "background_concepts":[],
         }
         if agent is None:
-            return build_quantitative_contract(topic, keywords, fallback_candidate)
+            return build_quantitative_contract(
+                topic, keywords, fallback_candidate,
+                source_topic=source_topic, source_keywords=source_keywords,
+            )
         prompt = f"""Design the minimum coherent quantitative management-study graph for the topic below.
 Return ONLY valid JSON with this schema:
 {{
@@ -532,11 +537,17 @@ Keywords: {keywords}"""
                 logical_node='structured_json',
             )
             parsed = agent._extract_json_from_text(raw)
-            contract = build_quantitative_contract(topic, keywords, parsed if isinstance(parsed, dict) else fallback_candidate)
+            contract = build_quantitative_contract(
+                topic, keywords, parsed if isinstance(parsed, dict) else fallback_candidate,
+                source_topic=source_topic, source_keywords=source_keywords,
+            )
             issues = validate_quantitative_contract(contract)
             if issues:
                 logging.warning("Adaptive quantitative design validation requires intent-preserving fallback: %s", issues[:3])
-                fallback = build_quantitative_contract(topic, keywords, fallback_candidate)
+                fallback = build_quantitative_contract(
+                    topic, keywords, fallback_candidate,
+                    source_topic=source_topic, source_keywords=source_keywords,
+                )
                 fallback_issues = validate_quantitative_contract(fallback)
                 if fallback_issues:
                     raise RuntimeError(f"quantitative_user_intent_unresolved:{fallback_issues[:3]}")
@@ -550,11 +561,17 @@ Keywords: {keywords}"""
                 candidate = parsed if isinstance(parsed, dict) else fallback_candidate
                 candidate = dict(candidate)
                 candidate["research_context"] = fallback_context
-                contract = build_quantitative_contract(topic, keywords, candidate)
+                contract = build_quantitative_contract(
+                    topic, keywords, candidate,
+                    source_topic=source_topic, source_keywords=source_keywords,
+                )
             return contract
         except Exception as exc:
             logging.warning("Quantitative adaptive study-design freeze fell back safely: %s", exc)
-            return build_quantitative_contract(topic, keywords, fallback_candidate)
+            return build_quantitative_contract(
+                topic, keywords, fallback_candidate,
+                source_topic=source_topic, source_keywords=source_keywords,
+            )
 
     def generate_empirical_data_robust(
         self,
@@ -2335,6 +2352,8 @@ Keywords: {keywords}"""
         *,
         topic="",
         keywords="",
+        internal_topic="",
+        internal_keywords=None,
         agent=None,
     ):
         if uploaded_files:
@@ -2358,9 +2377,20 @@ Keywords: {keywords}"""
                 metadata["quantitative_evidence_ledger"] = self.build_quantitative_evidence_ledger(metadata, user_md)
                 return None, user_md, tr(language, "actual_stats_parsed"), metadata
 
-        design_contract = self._derive_system_design(agent, topic, keywords)
+        design_topic = str(internal_topic or topic or "").strip()
+        design_keywords = internal_keywords if internal_keywords not in (None, "", []) else keywords
+        design_contract = self._derive_system_design(
+            agent, design_topic, design_keywords,
+            source_topic=topic, source_keywords=keywords,
+        )
         issues = validate_quantitative_contract(design_contract)
         if issues:
+            p1_identity_issues = [
+                issue for issue in issues
+                if issue.startswith("internal_english_") or issue.startswith("quantitative_construct_")
+            ]
+            if p1_identity_issues:
+                raise RuntimeError(f"quantitative_construct_identity_corruption:{p1_identity_issues[:4]}")
             raise RuntimeError(f"adaptive_quantitative_study_design_invalid:{issues[:4]}")
         quant = dict(design_contract.get("quantitative") or {})
         construct_items = [dict(item) for item in (quant.get("constructs") or []) if isinstance(item, dict)]

core/study_design_contract.py

@@ -28,7 +28,7 @@ def _clean(value: Any, maximum: int = 240) -> str:
 
 
 KEYWORD_POLICY_VERSION = "smartpaper-keyword-policy-v1"
-KEYWORD_SEPARATOR_POLICY = "SEMICOLON_NEWLINE_PIPE_PRIMARY_COMMA_FALLBACK"
+KEYWORD_SEPARATOR_POLICY = "UNICODE_SEMICOLON_NEWLINE_PIPE_PRIMARY_COMMA_FALLBACK"
 _KEYWORD_COUNT_MODES = {"UNSPECIFIED", "PREFERRED", "REQUIRED"}
 _KEYWORD_AUTHORITY_SOURCES = {"USER_EXPLICIT", "SYSTEM_GENERATED", "LEGACY"}
 
@@ -45,10 +45,10 @@ def parse_keyword_terms(value: Any, *, explicit_user: bool = False, diagnostics:
         raw = list(value)
     else:
         text = str(value or "")
-        if explicit_user and re.search(r"[;\n|]", text):
-            raw = re.split(r"[;\n|]+", text)
+        if explicit_user and re.search(r"[;；\n|]", text):
+            raw = re.split(r"[;；\n|]+", text)
         else:
-            raw = re.split(r"[,;\n|]+", text)
+            raw = re.split(r"[,，;；\n|]+", text)
     out: list[str] = []
     seen: set[str] = set()
     for item in raw:
@@ -406,7 +406,8 @@ def _append_relation(relations: list[dict[str, Any]], *, rtype: str, source: str
 
 
 def _reconcile_explicit_quantitative_intent(
-    *, topic: str, keywords: Any, constructs: list[dict[str, Any]], relations: list[dict[str, Any]], decisions: list[dict[str, Any]]
+    *, topic: str, keywords: Any, constructs: list[dict[str, Any]], relations: list[dict[str, Any]], decisions: list[dict[str, Any]],
+    source_topic: str | None = None, source_keywords: Any = None,
 ) -> dict[str, Any]:
     """Make explicit user design intent canonical rather than merely additive.
 
@@ -416,14 +417,27 @@ def _reconcile_explicit_quantitative_intent(
     assigned alternative moderators are advisory proposals and are removed
     before the canonical user relation is compiled.
     """
-    intent = build_user_research_intent(topic, keywords, mode="quantitative")
-    explicit = set(intent.get("explicit_relation_types") or [])
+    source_intent = build_user_research_intent(
+        topic if source_topic is None else source_topic,
+        keywords if source_keywords is None else source_keywords,
+        mode="quantitative",
+    )
+    identity_intent = build_user_research_intent(topic, keywords, mode="quantitative")
+    explicit = set(source_intent.get("explicit_relation_types") or [])
     if not explicit:
-        return intent
+        return source_intent
 
-    intent_text = " ; ".join([intent.get("source_topic") or ""] + list(intent.get("source_keywords") or []))
+    # Construct/hypothesis identity is compiled from the validated English
+    # projection, while the immutable source-language user intent remains the
+    # semantic authority recorded in the contract.
+    intent_text = " ; ".join(
+        [identity_intent.get("source_topic") or ""] + list(identity_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}]
+    labels += [
+        kw for kw in (identity_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 [])]
@@ -526,7 +540,7 @@ def _reconcile_explicit_quantitative_intent(
                         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
+    return source_intent
 
 def _stable_hash(payload: dict[str, Any]) -> str:
     encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
@@ -594,6 +608,8 @@ def build_quantitative_contract(
     *,
     source_type: str = "system_generated",
     policy_version: str = "adaptive-quantitative-v1",
+    source_topic: str | None = None,
+    source_keywords: Any = None,
 ) -> dict[str, Any]:
     """Normalize a model-proposed quantitative graph into deterministic authority."""
     raw = dict(candidate or {}) if isinstance(candidate, dict) else {}
@@ -602,12 +618,36 @@ def build_quantitative_contract(
         raw = _legacy_quant_candidate(list(raw_constructs), _clean(raw.get("research_context")))
         raw_constructs = raw["constructs"]
 
+    has_explicit_source_authority = source_topic is not None or source_keywords is not None
+    source_topic_value = topic if source_topic is None else source_topic
+    source_keywords_value = keywords if source_keywords is None else source_keywords
+    source_terms = parse_keyword_terms(source_keywords_value, explicit_user=True)
+    identity_terms = parse_keyword_terms(keywords, explicit_user=True)
+    identity_aliases: dict[str, str] = {}
+    if has_explicit_source_authority and source_terms and len(source_terms) == len(identity_terms):
+        identity_aliases = {
+            str(source).casefold(): str(target)
+            for source, target in zip(source_terms, identity_terms)
+            if str(source).strip() and str(target).strip()
+        }
+
+    def canonical_identity(value: Any) -> str:
+        label = _clean(value, 80)
+        return identity_aliases.get(label.casefold(), label)
+
     constructs: list[dict[str, Any]] = []
     label_to_id: dict[str, str] = {}
+    rejected: list[dict[str, Any]] = []
     for item in raw_constructs[:8]:
         if not isinstance(item, dict):
             continue
-        label = _clean(item.get("label") or item.get("name"), 80)
+        label = canonical_identity(item.get("label") or item.get("name"))
+        # A construct label containing a primary keyword separator is a malformed
+        # keyword blob, not a legitimate latent-variable identity. Reject it
+        # before relations/statistics can be compiled from it.
+        if re.search(r"[;；|\n]", label):
+            rejected.append({"reason": "keyword_blob_construct_rejected", "label": label})
+            continue
         if not label or label.casefold() in label_to_id:
             continue
         roles = _normalize_roles(item.get("roles") or item.get("role"))
@@ -627,17 +667,19 @@ def build_quantitative_contract(
     # Safe compatibility fallback; importantly this is a fallback, not the normal schema.
     if len(constructs) < 2:
         raw = _legacy_quant_candidate(["Focal Predictor", "Focal Mediator", "Focal Outcome"], _clean(raw.get("research_context")))
-        return build_quantitative_contract(topic, keywords, raw, source_type=source_type, policy_version=policy_version)
+        return build_quantitative_contract(
+            topic, keywords, raw, source_type=source_type, policy_version=policy_version,
+            source_topic=source_topic_value, source_keywords=source_keywords_value,
+        )
 
     by_id = {row["construct_id"]: row for row in constructs}
     def resolve(ref: Any) -> str:
-        text = _clean(ref, 80)
+        text = canonical_identity(ref)
         if text in by_id:
             return text
         return label_to_id.get(text.casefold(), "")
 
     relations: list[dict[str, Any]] = []
-    rejected: list[dict[str, Any]] = []
     for item in (raw.get("relations") or [])[:24]:
         if not isinstance(item, dict):
             continue
@@ -671,7 +713,8 @@ def build_quantitative_contract(
     # 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
+        topic=topic, keywords=keywords, constructs=constructs, relations=relations, decisions=rejected,
+        source_topic=source_topic_value, source_keywords=source_keywords_value,
     )
     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}
@@ -779,6 +822,14 @@ def build_quantitative_contract(
         "compiler_decisions": rejected,
         "policy_version": policy_version,
     }
+    if has_explicit_source_authority:
+        contract["internal_english_authority"] = {
+            "version": "quantitative-internal-english-authority-v1",
+            "title": re.sub(r"\s+", " ", str(topic or "")).strip(),
+            "keywords": identity_terms,
+            "source_keyword_count": len(list(user_intent.get("source_keywords") or [])),
+            "translation_only": True,
+        }
     return _finalize(contract)
 
 
@@ -804,6 +855,22 @@ def validate_quantitative_contract(contract: dict[str, Any]) -> list[str]:
     # 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 {})
+    internal_authority = dict(record.get("internal_english_authority") or {})
+    if internal_authority:
+        internal_title = str(internal_authority.get("title") or "")
+        internal_keywords = [str(x).strip() for x in (internal_authority.get("keywords") or []) if str(x).strip()]
+        source_terms = [str(x).strip() for x in (intent.get("source_keywords") or []) if str(x).strip()]
+        if source_terms and len(internal_keywords) != len(source_terms):
+            issues.append("internal_english_keyword_cardinality_mismatch")
+        foreign_script = re.compile(r"[\u3400-\u9fff\u3040-\u30ff\u0400-\u052f]")
+        if foreign_script.search(" ".join([internal_title] + internal_keywords)):
+            issues.append("internal_english_authority_language_contamination")
+        for row in constructs:
+            label = str(row.get("label") or "")
+            if re.search(r"[;；|\n]", label):
+                issues.append("quantitative_construct_keyword_blob")
+            if foreign_script.search(label):
+                issues.append("quantitative_construct_identity_language_contamination")
     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")