# -*- coding: utf-8 -*- """Adaptive study-design contracts shared by SmartPaper quantitative/qualitative paths. The LLM may propose a candidate design, but this module owns deterministic normalization, identity, validation, and scope semantics. It deliberately contains no provider/model routing logic. """ from __future__ import annotations import hashlib import json import re from typing import Any STUDY_DESIGN_CONTRACT_VERSION = "smartpaper-adaptive-study-design-contract-v1" _QUANT_ROLES = {"predictor", "mediator", "moderator", "outcome", "control", "covariate"} _QUANT_RELATIONS = {"direct_association", "moderation", "control_effect"} _QUAL_SOURCE_FAMILIES = { "interview", "focus_group", "observation", "document", "log", "artifact", "digital_trace", "field_note", "other", } def _clean(value: Any, maximum: int = 240) -> str: text = re.sub(r"\s+", " ", str(value or "")).strip(" \t\r\n,;:.-") return text[:maximum] KEYWORD_POLICY_VERSION = "smartpaper-keyword-policy-v1" KEYWORD_SEPARATOR_POLICY = "UNICODE_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: out.append(text) seen.add(key) return out[:16] USER_RESEARCH_INTENT_VERSION = "smartpaper-user-research-intent-v1" def _verbatim_keywords(value: Any) -> list[str]: """Preserve explicit user keyword identity/order under the resolved separator policy.""" return parse_keyword_terms(value, explicit_user=True) _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() 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): 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, "keyword_policy": keyword_policy, "keyword_diagnostics": keyword_diagnostics, "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]], source_topic: str | None = None, source_keywords: Any = None, ) -> dict[str, Any]: """Make explicit user design intent canonical rather than merely additive. R7-R2C-R5 restored omitted mediation/moderation relations, but an upstream model could still leave contradictory relations in place. S1-R1 closes that authority gap: when the user explicitly names a moderator, model- assigned alternative moderators are advisory proposals and are removed before the canonical user relation is compiled. """ 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 source_intent # 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 (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 [])] if "moderation" in explicit: moderator_ids = ids_for("moderator") # A phrase-bound moderator named by the user is canonical authority. # Model-assigned moderator roles/relations may not coexist when they # contradict that explicit identity. moderator_label = _phrase_bound_intent_label(intent_text, labels, "moderation") authoritative_named_moderator = bool(moderator_label) 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: if authoritative_named_moderator: # Remove only the conflicting moderator role. Other legitimate # roles (e.g. predictor/main-effect participation) are preserved. for row in constructs: if str(row.get("construct_id") or "") == moderator_id: continue roles = list(row.get("roles") or []) if "moderator" in roles: row["roles"] = [role for role in roles if role != "moderator"] removed = [] retained = [] for relation in relations: if relation.get("type") == "moderation" and str(relation.get("moderator") or "") != moderator_id: removed.append(dict(relation)) continue retained.append(relation) if removed: relations[:] = retained decisions.append({ "reason": "conflicting_model_moderation_removed_by_explicit_user_authority", "moderator": moderator_id, "removed_relation_ids": [str(row.get("relation_id") or "") for row in removed], }) 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 authoritative_named_moderator: # After removing contradictory model hypotheses, close any H-number # gap deterministically. This is identity normalization only; it # does not change relation semantics or create extra hypotheses. numbered = [ (idx, row) for idx, row in enumerate(relations) if row.get("numbered_hypothesis", True) and row.get("type") != "control_effect" ] numbered.sort(key=lambda item: ( int(item[1].get("hypothesis_order")) if str(item[1].get("hypothesis_order") or "").isdigit() else 10**6, item[0], )) for order, (_idx, row) in enumerate(numbered, start=1): row["hypothesis_order"] = order 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 source_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() def _finalize(contract: dict[str, Any]) -> dict[str, Any]: value = dict(contract) value.pop("contract_sha256", None) value.pop("contract_id", None) digest = _stable_hash(value) value["contract_id"] = digest[:20] value["contract_sha256"] = digest return value def _normalize_quant_relation_type(value: Any) -> str: text = _clean(value, 80).casefold().replace("-", "_").replace(" ", "_") aliases = { "association": "direct_association", "direct": "direct_association", "direct_effect": "direct_association", "direct_association": "direct_association", "path": "direct_association", "mediation_path": "direct_association", "moderator": "moderation", "moderation": "moderation", "interaction": "moderation", "interaction_effect": "moderation", "control": "control_effect", "control_effect": "control_effect", } return aliases.get(text, text) def _normalize_roles(value: Any) -> list[str]: raw = value if isinstance(value, (list, tuple)) else [value] out: list[str] = [] for item in raw: role = _clean(item, 40).casefold().replace(" ", "_") if role in _QUANT_ROLES and role not in out: out.append(role) return out def _legacy_quant_candidate(constructs: list[str], context: str = "") -> dict[str, Any]: values = [_clean(item, 80) for item in constructs if _clean(item, 80)] if len(values) != 3: values = ["Focal Predictor", "Focal Mediator", "Focal Outcome"] p, m, o = values return { "constructs": [ {"label": p, "roles": ["predictor"]}, {"label": m, "roles": ["mediator"]}, {"label": o, "roles": ["outcome"]}, ], "relations": [ {"source": p, "target": o, "type": "direct_association", "hypothesis_order": 1}, {"source": p, "target": m, "type": "direct_association", "hypothesis_order": 2}, {"source": m, "target": o, "type": "direct_association", "hypothesis_order": 3}, ], "indirect_effects": [{"path": [p, m, o], "kind": "simple", "numbered_hypothesis": False}], "research_context": context, "background_concepts": [], } def build_quantitative_contract( topic: str, keywords: Any, candidate: Any, *, source_type: str = "system_generated", policy_version: str = "adaptive-quantitative-v1", 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 {} raw_constructs = raw.get("constructs") or [] if raw_constructs and all(not isinstance(item, dict) for item in raw_constructs): raw = _legacy_quant_candidate(list(raw_constructs), _clean(raw.get("research_context"))) raw_constructs = raw["constructs"] 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 = 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")) if not roles: continue cid = f"C{len(constructs) + 1:03d}" row = { "construct_id": cid, "label": label, "roles": roles, "measurement_model": _clean(item.get("measurement_model") or "reflective", 40).casefold() or "reflective", "measurement_allowed": bool(item.get("measurement_allowed", True)), } constructs.append(row) label_to_id[label.casefold()] = cid # Safe compatibility fallback; importantly this is a fallback, not the normal schema. if len(constructs) < 2: raw = _legacy_quant_candidate(["Focal Predictor", "Focal Mediator", "Focal Outcome"], _clean(raw.get("research_context"))) return build_quantitative_contract( topic, keywords, raw, source_type=source_type, policy_version=policy_version, 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 = canonical_identity(ref) if text in by_id: return text return label_to_id.get(text.casefold(), "") relations: list[dict[str, Any]] = [] for item in (raw.get("relations") or [])[:24]: if not isinstance(item, dict): continue rtype = _normalize_quant_relation_type(item.get("type")) source = resolve(item.get("source")) target = resolve(item.get("target")) moderator = resolve(item.get("moderator")) if item.get("moderator") else "" if rtype not in _QUANT_RELATIONS or not source or not target or source == target: rejected.append({"reason": "invalid_or_unsupported_relation", "raw": {k: item.get(k) for k in ("type", "source", "target", "moderator")}}) continue if rtype == "moderation" and (not moderator or moderator in {source, target}): rejected.append({"reason": "moderation_requires_distinct_moderator", "raw": {k: item.get(k) for k in ("type", "source", "target", "moderator")}}) continue relation = { "relation_id": f"R{len(relations) + 1:03d}", "type": rtype, "source": source, "target": target, "moderator": moderator or None, "analysis_target": bool(item.get("analysis_target", True)), "numbered_hypothesis": bool(item.get("numbered_hypothesis", rtype != "control_effect")), } try: relation["hypothesis_order"] = int(item.get("hypothesis_order")) if item.get("hypothesis_order") is not None else None except Exception: relation["hypothesis_order"] = None relations.append(relation) # 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, 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} # If the interpreter omitted relations but roles make a parsimonious graph unambiguous, # compile a safe role-derived graph rather than silently returning an empty model. if not relations: predictors = [x["construct_id"] for x in constructs if "predictor" in x["roles"]] mediators = [x["construct_id"] for x in constructs if "mediator" in x["roles"]] outcomes = [x["construct_id"] for x in constructs if "outcome" in x["roles"]] moderators = [x["construct_id"] for x in constructs if "moderator" in x["roles"]] if predictors and outcomes: p, o = predictors[0], outcomes[0] if mediators: m = mediators[0] relations.extend([ {"relation_id":"R001","type":"direct_association","source":p,"target":o,"moderator":None,"analysis_target":True,"numbered_hypothesis":True,"hypothesis_order":1}, {"relation_id":"R002","type":"direct_association","source":p,"target":m,"moderator":None,"analysis_target":True,"numbered_hypothesis":True,"hypothesis_order":2}, {"relation_id":"R003","type":"direct_association","source":m,"target":o,"moderator":None,"analysis_target":True,"numbered_hypothesis":True,"hypothesis_order":3}, ]) else: relations.append({"relation_id":"R001","type":"direct_association","source":p,"target":o,"moderator":None,"analysis_target":True,"numbered_hypothesis":True,"hypothesis_order":1}) if moderators: relations.append({"relation_id":f"R{len(relations)+1:03d}","type":"moderation","source":p,"target":o,"moderator":moderators[0],"analysis_target":True,"numbered_hypothesis":True,"hypothesis_order":len(relations)+1}) direct_pairs = {(r["source"], r["target"]): r["relation_id"] for r in relations if r["type"] == "direct_association"} indirect_effects: list[dict[str, Any]] = [] raw_indirect = raw.get("indirect_effects") or [] for item in raw_indirect[:12]: if not isinstance(item, dict): continue path = [resolve(x) for x in (item.get("path") or [])] path = [x for x in path if x] if len(path) < 3 or any((a, b) not in direct_pairs for a, b in zip(path, path[1:])): rejected.append({"reason":"invalid_indirect_path","raw":{"path": item.get("path")}}) continue indirect_effects.append({ "effect_id": f"I{len(indirect_effects)+1:03d}", "path": path, "kind": _clean(item.get("kind") or ("simple" if len(path)==3 else "serial"), 30).casefold(), "numbered_hypothesis": bool(item.get("numbered_hypothesis", False)), }) if not indirect_effects: # Infer valid mediator chains from role identity and existing path edges. predictors = [x["construct_id"] for x in constructs if "predictor" in x["roles"]] mediators = [x["construct_id"] for x in constructs if "mediator" in x["roles"]] outcomes = [x["construct_id"] for x in constructs if "outcome" in x["roles"]] for p in predictors: for m in mediators: for o in outcomes: if (p, m) in direct_pairs and (m, o) in direct_pairs: indirect_effects.append({"effect_id":f"I{len(indirect_effects)+1:03d}","path":[p,m,o],"kind":"simple","numbered_hypothesis":False}) controls = [x["construct_id"] for x in constructs if set(x["roles"]) & {"control", "covariate"}] outcomes = [x["construct_id"] for x in constructs if "outcome" in x["roles"]] if not outcomes: targeted = {r["target"] for r in relations if r["type"] != "control_effect"} outcomes = sorted(targeted) analysis_plan = ["measurement_assessment"] if any(r["type"] == "direct_association" for r in relations): analysis_plan.append("direct_association") if indirect_effects: analysis_plan.append("indirect_effect") if any(r["type"] == "moderation" for r in relations): analysis_plan.extend(["interaction_test", "simple_slopes"]) analysis_plan.append("robustness") # Legacy profile is used only to preserve exact prior H ordering and validators. role_map = {role: [x for x in constructs if role in x["roles"]] for role in ("predictor","mediator","outcome")} legacy_profile = ( len(constructs) == 3 and all(len(role_map[r]) == 1 for r in role_map) and not any("moderator" in x["roles"] or set(x["roles"]) & {"control","covariate"} for x in constructs) and len([r for r in relations if r["type"] == "direct_association"]) == 3 ) unsupported = _keywords(raw.get("background_concepts") or raw.get("unsupported_seed_concepts") or []) context = _clean(raw.get("research_context") or "respondents with experience relevant to the focal management phenomenon", 240) contract = { "version": STUDY_DESIGN_CONTRACT_VERSION, "mode": "quantitative", "source_type": _clean(source_type, 40) or "system_generated", "source_topic": 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", "evidence_strength": { "causal_identification": False, "allowed_inference": ["association", "estimated_indirect_association", "interaction_association"], "forbidden_inference": ["causal_effect", "causal_mediation_identification"], }, "context": {"research_context": context}, "quantitative": { "constructs": constructs, "relations": relations, "indirect_effects": indirect_effects, "controls": controls, "outcomes": outcomes, "analysis_plan": analysis_plan, "legacy_three_construct_profile": legacy_profile, }, "qualitative": None, "unsupported_seed_concepts": unsupported, "compiler_decisions": rejected, "policy_version": policy_version, } 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) def validate_quantitative_contract(contract: dict[str, Any]) -> list[str]: issues: list[str] = [] record = dict(contract or {}) if record.get("version") != STUDY_DESIGN_CONTRACT_VERSION or record.get("mode") != "quantitative": return ["invalid_contract_identity"] quant = dict(record.get("quantitative") or {}) constructs = [x for x in (quant.get("constructs") or []) if isinstance(x, dict)] ids = {x.get("construct_id") for x in constructs} if len(ids) != len(constructs) or len(constructs) < 2: issues.append("invalid_construct_identity") 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 if r.get("source") not in ids or r.get("target") not in ids: issues.append("relation_references_unknown_construct") if r.get("type") == "moderation" and r.get("moderator") not in ids: issues.append("moderation_references_unknown_moderator") # 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") 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)) def compile_quantitative_hypothesis_ledger(contract: dict[str, Any], *, version: str) -> dict[str, Any]: quant = dict((contract or {}).get("quantitative") or {}) constructs = {x.get("construct_id"): x for x in (quant.get("constructs") or []) if isinstance(x, dict)} rels = [dict(r) for r in (quant.get("relations") or []) if isinstance(r, dict) and r.get("numbered_hypothesis")] rels.sort(key=lambda r: (r.get("hypothesis_order") is None, r.get("hypothesis_order") or 999, r.get("relation_id") or "")) items: list[dict[str, Any]] = [] for r in rels: source = constructs.get(r.get("source"), {}).get("label") target = constructs.get(r.get("target"), {}).get("label") moderator = constructs.get(r.get("moderator"), {}).get("label") if r.get("moderator") else None if not source or not target: continue row = { "hypothesis_id": f"H{len(items)+1}", "predictor": source, "outcome": target, "relation": "moderation" if r.get("type") == "moderation" else "association", "relation_id": r.get("relation_id"), } if moderator: row["moderator"] = moderator items.append(row) indirect_rows = [] for effect in quant.get("indirect_effects") or []: if not isinstance(effect, dict): continue labels = [constructs.get(cid, {}).get("label") for cid in effect.get("path") or []] labels = [x for x in labels if x] if len(labels) >= 3: indirect_rows.append({ "effect_id": effect.get("effect_id"), "path": labels, "predictor": labels[0], "mediator": " -> ".join(labels[1:-1]), "outcome": labels[-1], "numbered_hypothesis": bool(effect.get("numbered_hypothesis", False)), "relation": "estimated_indirect_association", }) ledger = {"version": version, "items": items, "indirect_analyses": indirect_rows} # Exact compatibility key for the historical single-mediator profile. if len(indirect_rows) == 1 and len(indirect_rows[0].get("path") or []) == 3: p = indirect_rows[0] ledger["indirect_analysis"] = { "predictor": p["path"][0], "mediator": p["path"][1], "outcome": p["path"][2], "numbered_hypothesis": p["numbered_hypothesis"], "relation": p["relation"], } return ledger def build_quantitative_front_matter_contract(contract: dict[str, Any], *, version: str) -> dict[str, Any]: quant = dict((contract or {}).get("quantitative") or {}) items = [dict(x) for x in (quant.get("constructs") or []) if isinstance(x, dict)] # Preserve legacy role map where a role is singular, while adding an extensible item list. role_map: dict[str, str] = {} for role in ("predictor", "mediator", "moderator", "outcome", "control", "covariate"): values = [x.get("label") for x in items if role in (x.get("roles") or []) and x.get("label")] if len(values) == 1: role_map[role] = values[0] construct_roles = [r for r in ("predictor","mediator","moderator","outcome","control","covariate") if r in role_map] return { "version": version, "policy": "adaptive_study_design_frozen_semantic_authority", "source_topic": str((contract or {}).get("source_topic") or ""), "source_keywords": list((contract or {}).get("source_keywords") or []), # S1-R1: compatibility front-matter metadata must carry the complete # R5/R6 user Research Intent Authority. Without it a degraded final # reconciliation can silently synthesize a different title/keyword set. "user_research_intent": dict((contract or {}).get("user_research_intent") or {}), "construct_roles": construct_roles, "constructs": role_map, "construct_items": items, "relations": list(quant.get("relations") or []), "unsupported_seed_concepts": list((contract or {}).get("unsupported_seed_concepts") or []), "study_design_contract_sha256": str((contract or {}).get("contract_sha256") or ""), } def build_qualitative_contract(topic: str, keywords: Any, metadata: dict[str, Any], *, policy_version: str = "adaptive-qualitative-v1") -> dict[str, Any]: meta = dict(metadata or {}) facts = dict(meta.get("study_facts") or {}) 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) folded = label.casefold() family = ( "interview" if "interview" in folded else "focus_group" if "focus" in folded and "group" in folded else "observation" if "observ" in folded else "log" if "log" in folded else "document" if any(x in folded for x in ("document","archiv","record")) else "digital_trace" if any(x in folded for x in ("digital trace","clickstream","platform trace")) else "field_note" if "field note" in folded else "artifact" if "artifact" in folded else "other" ) sources.append({"source_id":f"D{index:03d}","label":label,"family":family,"semantic_descriptor":label if family=="other" else ""}) actors = [] seen = set() for row in facts.get("data_source_roles") or []: if not isinstance(row, dict): continue label = _clean(row.get("actor_role"), 100) if label and label.casefold() not in seen: actors.append({"actor_id":f"A{len(actors)+1:03d}","label":label}); seen.add(label.casefold()) themes = [] for item in meta.get("themes") or []: if not isinstance(item, dict): continue themes.append({ "theme_id": _clean(item.get("theme_id"), 40), "title": _clean(item.get("title"), 160), "stages": [_clean(x,80) for x in (item.get("stages") or []) if _clean(x,80)], "has_deviant_evidence": bool(item.get("has_deviant_evidence")), }) design = _clean(facts.get("analytic_design") or facts.get("analysis_method"), 120) # 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": 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", "evidence_strength": {"causal_identification":False,"allowed_inference":["theme","process_pattern","interpretive_mechanism"],"forbidden_inference":["causal_effect"]}, "context": {"research_setting": _clean(facts.get("research_setting"),240)}, "quantitative": None, "qualitative": { "phenomenon": _clean(topic,300), "organizational_puzzle": "", "research_questions": [], "design_tradition": design or "Iterative thematic/process analysis", "units_of_analysis": [{"unit_id":"U001","label":case_structure or "study participants/organizational setting","level":"other"}], "actor_groups": actors, "data_sources": sources, "sampling": {"participant_count": facts.get("participant_count"), "interview_count": facts.get("interview_count")}, "temporal_scope": {"description": temporal or "Not specified"}, "analysis_strategy": [design] if design else [], "themes": themes, "process_model": [], "trustworthiness_procedures": list(facts.get("trustworthiness_procedures") or []), "ethics_procedures": list(facts.get("ethics_procedures") or []), "reflexivity_procedures": list(facts.get("reflexivity_procedures") or []), }, "unsupported_seed_concepts": [], "compiler_decisions": [], "policy_version": policy_version, } return _finalize(contract) def _study_owned_sentence(sentence: str) -> bool: text = str(sentence or "") if re.search(r"\bCITE-\d+\b", text, flags=re.I): return False if re.search(r"\((?=[^()\n]*(?:19|20)\d{2})[^()\n]+\)", text): return False if re.search(r"(?i)\b(?:prior|previous|existing|earlier)\s+(?:research|stud(?:y|ies)|literature)\b", text): return False return bool(re.search(r"(?i)\b(?:we|our\s+study|this\s+study|the\s+present\s+study|the\s+study|our\s+analysis|our\s+results|our\s+findings|we\s+(?:examine|investigate|test|analy[sz]e|explore|study))\b", text)) def inspect_study_design_scope(text: str, contract: dict[str, Any], *, section_name: str = "") -> list[dict[str, Any]]: """High-confidence whole-manuscript scope guard for both modes.""" value = str(text or "") rows: list[dict[str, Any]] = [] mode = str((contract or {}).get("mode") or "") sentences = [s.strip() for s in re.split(r"(?<=[.!?])\s+|\n+", value) if s.strip()] if mode == "quantitative": quant = dict((contract or {}).get("quantitative") or {}) constructs = [x for x in (quant.get("constructs") or []) if isinstance(x, dict)] allowed_labels = {str(x.get("label") or "").casefold() for x in constructs if x.get("label")} has_moderation = any(r.get("type") == "moderation" for r in (quant.get("relations") or []) if isinstance(r, dict)) has_indirect = bool(quant.get("indirect_effects")) unsupported = [str(x) for x in (contract or {}).get("unsupported_seed_concepts") or [] if str(x).strip()] for sentence in sentences: if not _study_owned_sentence(sentence): continue folded = sentence.casefold() if re.search(r"\bmoderat(?:e|es|ed|ing|or|ion)\b|\binteraction\s+(?:effect|term)\b", folded) and not has_moderation: rows.append({"type":"unsupported_moderation_claim","preview":sentence[:260],"reason":"The frozen study-design graph contains no moderation/interaction analysis target."}) if re.search(r"\bmediat(?:e|es|ed|ing|or|ion)\b|\bindirect\s+(?:effect|association)\b", folded) and not has_indirect: rows.append({"type":"unsupported_mediation_claim","preview":sentence[:260],"reason":"The frozen study-design graph contains no indirect-effect analysis target."}) for concept in unsupported: if concept.casefold() in folded and re.search(r"(?i)\b(?:examine|investigate|test|model|measure|assess|estimate)\b", sentence): rows.append({"type":"unsupported_focal_construct","construct":concept,"preview":sentence[:260],"reason":"The concept is background-only under the frozen study-design contract."}) elif mode == "qualitative": qual = dict((contract or {}).get("qualitative") or {}) design = str(qual.get("design_tradition") or "").casefold() temporal = str(dict(qual.get("temporal_scope") or {}).get("description") or "").casefold() families = {str(x.get("family") or "") for x in (qual.get("data_sources") or []) if isinstance(x, dict)} unit_label = " ".join(str(x.get("label") or "") for x in (qual.get("units_of_analysis") or []) if isinstance(x, dict)).casefold() for sentence in sentences: if not _study_owned_sentence(sentence): continue folded = sentence.casefold() if "longitudinal" in folded and "longitudinal" not in temporal and "longitudinal" not in design: rows.append({"type":"unsupported_temporal_scope","preview":sentence[:260],"reason":"The frozen qualitative design does not establish longitudinal evidence."}) if "ethnograph" in folded and "ethnograph" not in design: rows.append({"type":"unsupported_design_tradition","preview":sentence[:260],"reason":"Ethnography is not the frozen qualitative design tradition."}) if "grounded theory" in folded and "grounded theory" not in design: rows.append({"type":"unsupported_design_tradition","preview":sentence[:260],"reason":"Grounded theory is not the frozen qualitative design tradition."}) if "triangulat" in folded and len({x for x in families if x}) < 2: rows.append({"type":"unsupported_data_source_claim","preview":sentence[:260],"reason":"Triangulation requires at least two frozen evidence-source families."}) if re.search(r"\b(?:multi|multiple)[- ]case\b|\bcomparative case\b", folded) and not any(x in (design + " " + unit_label) for x in ("comparative", "multiple", "multi-case")): rows.append({"type":"unsupported_unit_of_analysis","preview":sentence[:260],"reason":"The frozen qualitative contract does not establish a multi-case comparison."}) return rows[:8]