Blame for core/study_design_contract.py

# -*- 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]