Spaces:
Running
Running
| # -*- 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] | |