P0-R7-D-R2 research-material-only reliability closure

.streamlit/config.toml

@@ -3,7 +3,7 @@ enableStaticServing = true
 enableCORS = false
 enableXsrfProtection = false
 enableWebsocketCompression = false
-maxUploadSize = 1024
+maxUploadSize = 10
 
 [browser]
 gatherUsageStats = false

app.py

@@ -39,7 +39,11 @@ from core.advanced_method_registry import (
     reconcile_method_authority, resolve_generation_preflight,
 )
 from smartpaper_restore_bridge import edge_bridge
-from utils.file_parser import extract_file_content, validate_file_parseable
+from utils.file_parser import (
+    extract_research_material_content,
+    research_material_text_fingerprint,
+    validate_research_material_template,
+)
 
 st.set_page_config(
     page_title="Article Agent",
@@ -507,14 +511,18 @@ def _save_uploaded(items, inputs_dir: Path, prefix: str) -> list[str]:
 
 def _research_material_upload_policy() -> dict:
     policy = dict(APP_CONFIG.get("research_material_upload_policy") or {})
-    reference_limit = int(policy.get("reference_file_limit") or 0)
     research_limit = int(policy.get("research_material_file_limit") or 0)
-    if reference_limit <= 0 or research_limit <= 0:
+    if research_limit <= 0:
         raise RuntimeError("research_material_upload_policy_invalid")
     return {
         **policy,
-        "reference_file_limit": reference_limit,
         "research_material_file_limit": research_limit,
+        "research_material_file_max_bytes": max(
+            1024, int(policy.get("research_material_file_max_bytes") or 0)
+        ),
+        "research_material_total_max_bytes": max(
+            1024, int(policy.get("research_material_total_max_bytes") or 0)
+        ),
         "qualitative_min_combined_chars": max(
             1, int(policy.get("qualitative_min_combined_chars") or 1)
         ),
@@ -537,6 +545,7 @@ def _uploaded_file_identity(item, occurrence: int = 0) -> str:
 
 
 def _uploaded_material_entries() -> list[dict]:
+    """Return research-material-only upload entries for the current design."""
     uploads = list(st.session_state.get("up_research_materials") or [])
     occurrences: dict[tuple[str, str, str], int] = {}
     entries: list[dict] = []
@@ -547,23 +556,14 @@ def _uploaded_material_entries() -> list[dict]:
         signature = (name, size, mime)
         occurrence = occurrences.get(signature, 0)
         occurrences[signature] = occurrence + 1
-        identity = _uploaded_file_identity(item, occurrence)
-        key = f"sp_material_role_{identity}"
-        ext = Path(name).suffix.lower()
-        default_role = "research" if ext in {".txt", ".csv", ".xls", ".xlsx"} else "reference"
-        if key not in st.session_state:
-            st.session_state[key] = default_role
-        role = str(st.session_state.get(key) or default_role).strip().lower()
-        if role not in {"reference", "research"}:
-            role = default_role
         entries.append(
             {
                 "item": item,
-                "identity": identity,
-                "state_key": key,
+                "identity": _uploaded_file_identity(item, occurrence),
                 "name": name,
-                "extension": ext,
-                "role": role,
+                "extension": Path(name).suffix.lower(),
+                "size_bytes": int(getattr(item, "size", 0) or 0),
+                "role": "research",
             }
         )
     return entries
@@ -571,81 +571,165 @@ def _uploaded_material_entries() -> list[dict]:
 
 def _material_role_counts(entries: list[dict] | None = None) -> dict[str, int]:
     entries = list(entries if entries is not None else _uploaded_material_entries())
-    return {
-        "reference": sum(1 for entry in entries if entry.get("role") == "reference"),
-        "research": sum(1 for entry in entries if entry.get("role") == "research"),
-    }
+    # Keep the legacy reference key only for backward diagnostic compatibility.
+    # New-task UI exposes research materials only.
+    return {"reference": 0, "research": len(entries)}
 
 
 def _material_limits_ok(entries: list[dict] | None = None) -> tuple[bool, dict[str, int], dict]:
     entries = list(entries if entries is not None else _uploaded_material_entries())
     counts = _material_role_counts(entries)
     policy = _research_material_upload_policy()
-    ok = (
-        counts["reference"] <= policy["reference_file_limit"]
-        and counts["research"] <= policy["research_material_file_limit"]
-    )
+    ok = counts["research"] <= policy["research_material_file_limit"]
     return ok, counts, policy
 
 
-def _explicit_material_bucket(method_family: str, role: str) -> str:
+def _research_material_extension_contract(method_family: str) -> tuple[str, ...]:
+    capability = method_capability(method_family)
+    extensions = tuple(
+        str(item or "").strip().casefold().lstrip(".")
+        for item in (capability.get("research_material_extensions") or ())
+        if str(item or "").strip()
+    )
+    if len(extensions) != 1 or extensions[0] not in {"txt", "csv"}:
+        raise RuntimeError("method_research_material_extension_unsupported")
+    return extensions
+
+
+def _explicit_material_bucket(method_family: str, role: str = "research") -> str:
+    if str(role or "research").strip().casefold() != "research":
+        raise RuntimeError("reference_upload_not_supported")
     capability = method_capability(method_family)
     material_routing = dict(capability.get("material_routing") or {})
-    route_key = "user_reference" if role == "reference" else "user_research"
-    bucket = str(material_routing.get(route_key) or "").strip().casefold()
-    if bucket not in {"reference", "data", "interview"}:
+    bucket = str(material_routing.get("user_research") or "").strip().casefold()
+    if bucket not in {"data", "interview"}:
         raise RuntimeError("method_material_routing_unsupported")
     return bucket
 
 
+def _canonical_user_keywords(value: object) -> str:
+    """Normalize all supported user separators while preserving phrase spaces."""
+    terms = []
+    seen = set()
+    for raw in re.split(r"[;；,，\n|]+", str(value or "")):
+        term = re.sub(r"\s+", " ", raw).strip()
+        key = term.casefold()
+        if not term or key in seen:
+            continue
+        terms.append(term)
+        seen.add(key)
+    return "; ".join(terms)
+
+
+def _template_content_fingerprints(template_profile: str) -> set[str]:
+    out: set[str] = set()
+    for language_code in LANGUAGE_OPTIONS.values():
+        content = str(get_research_material_template(language_code, template_profile).get("content") or "")
+        if content.strip():
+            out.add(research_material_text_fingerprint(content))
+    return out
+
+
 def _material_preflight_report(method_family: str, entries: list[dict] | None = None) -> dict:
-    """Validate every selected file before task creation/quota reservation."""
+    """Validate research files and capacity before task creation/quota reservation."""
     entries = list(entries if entries is not None else _uploaded_material_entries())
     limits_ok, counts, policy = _material_limits_ok(entries)
+    capability = method_capability(method_family)
+    template_profile = str(capability.get("template_profile") or "").strip()
+    expected_extensions = _research_material_extension_contract(method_family)
+    research_bucket = _explicit_material_bucket(method_family)
+    bundle_limit = int(
+        policy["quantitative_bundle_max_chars"]
+        if research_bucket == "data"
+        else policy["qualitative_bundle_max_chars"]
+    )
+    total_bytes = sum(max(0, int(entry.get("size_bytes") or 0)) for entry in entries)
     report = {
         "ok": bool(limits_ok),
         "counts": counts,
         "policy": policy,
         "failures": [],
         "research_chars": 0,
+        "bundle_chars": 0,
+        "total_bytes": total_bytes,
+        "template_profile": template_profile,
+        "expected_extensions": expected_extensions,
+        "bundle_limit_chars": bundle_limit,
     }
     if not limits_ok:
         return report
+    if total_bytes > int(policy["research_material_total_max_bytes"]):
+        report["ok"] = False
+        report["total_size_exceeded"] = True
+        return report
 
     cache = st.session_state.get("sp_material_parse_cache")
     if not isinstance(cache, dict):
         cache = {}
         st.session_state["sp_material_parse_cache"] = cache
 
+    template_fingerprints = _template_content_fingerprints(template_profile)
     live_cache_keys = set()
+    successful_entries = 0
     for entry in entries:
         item = entry["item"]
-        role = str(entry["role"])
-        cache_key = f"{entry['identity']}|{role}"
+        ext = str(entry.get("extension") or "").casefold().lstrip(".")
+        cache_key = f"{entry['identity']}|{template_profile}|r2"
         live_cache_keys.add(cache_key)
-        result = cache.get(cache_key)
-        if not isinstance(result, dict):
-            result = validate_file_parseable(item)
-            cache[cache_key] = dict(result)
+        size_bytes = max(0, int(entry.get("size_bytes") or 0))
+        if size_bytes > int(policy["research_material_file_max_bytes"]):
+            result = {
+                "ok": False,
+                "filename": entry["name"],
+                "text_chars": 0,
+                "byte_size": size_bytes,
+                "error": "research_material_file_too_large",
+            }
+        elif ext not in expected_extensions:
+            result = {
+                "ok": False,
+                "filename": entry["name"],
+                "text_chars": 0,
+                "byte_size": size_bytes,
+                "error": f"template_extension_mismatch:expected={expected_extensions[0]}",
+            }
+        else:
+            result = cache.get(cache_key)
+            if not isinstance(result, dict):
+                result = validate_research_material_template(
+                    item,
+                    template_profile,
+                    max_chars=None,
+                    max_bytes=int(policy["research_material_file_max_bytes"]),
+                    disallowed_content_fingerprints=template_fingerprints,
+                )
+                cache[cache_key] = dict(result)
         if not bool(result.get("ok")):
             report["failures"].append(
                 {
                     "name": entry["name"],
-                    "role": role,
+                    "role": "research",
                     "error": str(result.get("error") or "parse_failed"),
                 }
             )
-        elif role == "research":
-            report["research_chars"] += int(result.get("text_chars") or 0)
+        else:
+            chars = int(result.get("text_chars") or 0)
+            report["research_chars"] += chars
+            report["bundle_chars"] += len(f"### Source: {entry['name']}\n") + chars
+            successful_entries += 1
+
+    if successful_entries > 1:
+        report["bundle_chars"] += 2 * (successful_entries - 1)
 
-    # Drop stale cache entries when files are removed or roles change.
     st.session_state["sp_material_parse_cache"] = {
         key: value for key, value in cache.items() if key in live_cache_keys
     }
 
     if report["failures"]:
         report["ok"] = False
-    research_bucket = _explicit_material_bucket(method_family, "research")
+    if report["bundle_chars"] > bundle_limit:
+        report["ok"] = False
+        report["bundle_capacity_exceeded"] = True
     if (
         report["ok"]
         and counts["research"] > 0
@@ -656,10 +740,8 @@ def _material_preflight_report(method_family: str, entries: list[dict] | None =
         report["content_insufficient"] = True
     return report
 
-
-def _material_bundle_text(entries: list[dict], research_bucket: str) -> str:
-    research_entries = [entry for entry in entries if entry.get("role") == "research"]
-    if not research_entries:
+def _material_bundle_text(entries: list[dict], research_bucket: str, template_profile: str) -> str:
+    if not entries:
         return ""
     policy = _research_material_upload_policy()
     if research_bucket == "data":
@@ -668,12 +750,19 @@ def _material_bundle_text(entries: list[dict], research_bucket: str) -> str:
         total_budget = int(policy["qualitative_bundle_max_chars"])
     else:
         raise RuntimeError("method_material_routing_unsupported")
-    per_file_budget = max(1, total_budget // len(research_entries))
     sections = []
-    for entry in research_entries:
-        text = extract_file_content(entry["item"], max_chars=per_file_budget)
+    for entry in entries:
+        text = extract_research_material_content(
+            entry["item"],
+            template_profile,
+            max_chars=total_budget,
+            max_bytes=int(policy["research_material_file_max_bytes"]),
+        )
         sections.append(f"### Source: {entry['name']}\n{text.strip()}")
-    return "\n\n".join(sections).strip()[:total_budget]
+    bundle = "\n\n".join(sections).strip()
+    if len(bundle) > total_budget:
+        raise RuntimeError("research_material_bundle_capacity_exceeded")
+    return bundle
 
 def _uploaded_material_descriptors() -> list[dict[str, str]]:
     out: list[dict[str, str]] = []
@@ -697,7 +786,7 @@ def _method_intent_signature() -> str:
         "topic": _normalized_research_intent_text(
             st.session_state.get("sp_topic") or DEFAULT_TOPIC
         ),
-        "keywords": _normalized_research_intent_text(
+        "keywords": _canonical_user_keywords(
             st.session_state.get("sp_keywords") or DEFAULT_KEYWORDS
         ),
     }
@@ -770,7 +859,7 @@ def _current_method_preflight(*, confirmed_method_family: str | None = None) ->
         )
     preflight = resolve_generation_preflight(
         str(st.session_state.get("sp_topic") or DEFAULT_TOPIC),
-        str(st.session_state.get("sp_keywords") or DEFAULT_KEYWORDS),
+        _canonical_user_keywords(st.session_state.get("sp_keywords") or DEFAULT_KEYWORDS),
         materials=_uploaded_material_descriptors(),
         confirmed_method_family=confirmed,
     )
@@ -779,11 +868,11 @@ def _current_method_preflight(*, confirmed_method_family: str | None = None) ->
 def _classify_and_save_research_materials(
     method_family: str, inputs_dir: Path
 ) -> tuple[list[str], list[str], list[str], str, list[dict]]:
-    """Persist all approved files while preserving the user's per-file role authority.
+    """Persist template-compatible research materials for the selected design.
 
-    Research materials are combined into one deterministic text bundle so the
-    frozen GenerationRunner can continue using its proven single empirical-input
-    seam without silently dropping files 2..N.
+    New tasks no longer accept uploaded reference literature.  Research files are
+    combined into one deterministic text bundle so the frozen GenerationRunner
+    continues using its proven single empirical-input seam.
     """
     entries = _uploaded_material_entries()
     if not entries:
@@ -793,55 +882,36 @@ def _classify_and_save_research_materials(
     if not report.get("ok"):
         raise RuntimeError("research_material_preflight_failed")
 
-    reference_items = [entry["item"] for entry in entries if entry.get("role") == "reference"]
-    research_entries = [entry for entry in entries if entry.get("role") == "research"]
-
-    refs = _save_uploaded(reference_items, inputs_dir, "ref")
     original_research = _save_uploaded(
-        [entry["item"] for entry in research_entries], inputs_dir, "research"
+        [entry["item"] for entry in entries], inputs_dir, "research"
     )
 
     data: list[str] = []
     interviews: list[str] = []
     capability = method_capability(method_family)
-    material_routing = dict(capability.get("material_routing") or {})
-    reference_bucket = str(material_routing.get("user_reference") or "").strip().casefold()
-    research_bucket = str(material_routing.get("user_research") or "").strip().casefold()
-    if reference_bucket != "reference" or research_bucket not in {"data", "interview"}:
-        raise RuntimeError("method_material_routing_unsupported")
-    if research_entries:
-        bundle_text = _material_bundle_text(research_entries, research_bucket)
-        if not bundle_text:
-            raise RuntimeError("research_material_bundle_empty")
-        bundle_name = "research_materials_bundle.txt"
-        bundle_path = inputs_dir / bundle_name
-        bundle_path.write_text(bundle_text, encoding="utf-8")
-        bundle_rel = str(Path("inputs") / bundle_name)
-        if research_bucket == "data":
-            data = [bundle_rel]
-        else:
-            interviews = [bundle_rel]
-
-    role_manifest = []
-    research_index = 0
-    reference_index = 0
-    for entry in entries:
-        role = str(entry.get("role") or "")
-        if role == "reference":
-            saved_path = refs[reference_index]
-            reference_index += 1
-        else:
-            saved_path = original_research[research_index]
-            research_index += 1
-        role_manifest.append(
-            {
-                "name": entry["name"],
-                "role": role,
-                "saved_path": saved_path,
-            }
-        )
+    research_bucket = _explicit_material_bucket(method_family)
+    template_profile = str(capability.get("template_profile") or "").strip()
+    bundle_text = _material_bundle_text(entries, research_bucket, template_profile)
+    if not bundle_text:
+        raise RuntimeError("research_material_bundle_empty")
+    bundle_name = "research_materials_bundle.txt"
+    bundle_path = inputs_dir / bundle_name
+    bundle_path.write_text(bundle_text, encoding="utf-8")
+    bundle_rel = str(Path("inputs") / bundle_name)
+    if research_bucket == "data":
+        data = [bundle_rel]
+    else:
+        interviews = [bundle_rel]
 
-    return refs, data, interviews, "per_file_v1", role_manifest
+    role_manifest = [
+        {
+            "name": entry["name"],
+            "role": "research",
+            "saved_path": original_research[index],
+        }
+        for index, entry in enumerate(entries)
+    ]
+    return [], data, interviews, "research_only_template_v2", role_manifest
 
 def _persist_task_request(
     task: dict,
@@ -881,7 +951,7 @@ def _persist_task_request(
         "target_lang": get_target_language(lang_code),
         "output_suffix": get_output_suffix(lang_code),
         "topic": str(st.session_state.get("sp_topic") or DEFAULT_TOPIC),
-        "keywords": str(st.session_state.get("sp_keywords") or DEFAULT_KEYWORDS),
+        "keywords": _canonical_user_keywords(st.session_state.get("sp_keywords") or DEFAULT_KEYWORDS),
         "target_words": int(st.session_state.get("sp_words") or APP_CONFIG["target_words_default"]),
         # High-entropy, display-only token. It is part of the immutable task
         # request/hash, but it grants no Worker/D1 authority.
@@ -1157,8 +1227,8 @@ def _process_pending_launch(lang_code: str) -> None:
 
 inject_wechat_browser_notice()
 
-DEFAULT_TOPIC = "Generative Artificial Intelligence in Ubiquitous Learning: Evaluating a Chatbot-based Recommendation Engine for Personalized and Context-aware Education"
-DEFAULT_KEYWORDS = "Ubiquitous computing; Context awareness; Deep learning; Personalized educational; Educational technologies; HCI; Recommendation system; LLMs; Generative AI; ChatGPT; Prompting engineering; Few-shot prompting"
+DEFAULT_TOPIC = "Generative Artificial Intelligence in Ubiquitous Learning"
+DEFAULT_KEYWORDS = "Ubiquitous computing; Context awareness; Deep learning; Personalized education; Educational technologies; Recommendation system;"
 
 def _current_language_code() -> str:
     choice = st.session_state.get("interface_language_choice")
@@ -1778,52 +1848,9 @@ def render_workspace_controls(user):
         key="sp_keywords",
     )
 
-    # U1-R2 unified research workspace: academic intent/materials first, then
-    # product/length controls. Legacy AMJ/ASQ remains internal policy routing.
-    st.file_uploader(
-        tr(lang_code, "research_materials_label"),
-        type=["pdf", "txt", "doc", "docx", "csv", "xls", "xlsx"],
-        accept_multiple_files=True,
-        key="up_research_materials",
-        disabled=active,
-        help=tr(lang_code, "research_materials_help"),
-    )
-
-    uploaded_materials = list(st.session_state.get("up_research_materials") or [])
-    material_entries = _uploaded_material_entries()
-    if uploaded_materials and not active:
-        for entry in material_entries:
-            st.radio(
-                entry["name"],
-                ["reference", "research"],
-                format_func=lambda value: tr(lang_code, f"material_use_{value}"),
-                horizontal=True,
-                key=entry["state_key"],
-            )
-        # Re-read after widgets because Streamlit may have changed role state.
-        material_entries = _uploaded_material_entries()
-        limits_ok, material_counts, material_policy = _material_limits_ok(material_entries)
-        if material_counts["reference"] > material_policy["reference_file_limit"]:
-            st.error(
-                tr(
-                    lang_code,
-                    "material_reference_limit_exceeded",
-                    limit=material_policy["reference_file_limit"],
-                )
-            )
-        if material_counts["research"] > material_policy["research_material_file_limit"]:
-            st.error(
-                tr(
-                    lang_code,
-                    "material_research_limit_exceeded",
-                    limit=material_policy["research_material_file_limit"],
-                )
-            )
-    else:
-        limits_ok = True
-        material_counts = {"reference": 0, "research": 0}
-        material_policy = _research_material_upload_policy()
-
+    # P0-R7-E1 simplified workspace: Research Design -> Recommended Template ->
+    # Research Materials.  New tasks accept empirical materials only; reference
+    # literature upload is intentionally removed from the product surface.
     if active:
         active_family = _task_method_family(task)
         active_capability = method_capability(active_family)
@@ -1862,29 +1889,6 @@ def render_workspace_controls(user):
     if active:
         st.markdown(f"**{tr(lang_code, 'detected_method_label')}:** {method_label}")
 
-    material_gate = {"ok": True, "failures": []}
-    if not active and limits_ok:
-        material_gate = _material_preflight_report(method_family, material_entries)
-        for failure in material_gate.get("failures") or []:
-            role = str(failure.get("role") or "reference")
-            formats_key = (
-                "material_reference_convert_formats"
-                if role == "reference"
-                else "material_research_convert_formats"
-            )
-            st.error(
-                tr(
-                    lang_code,
-                    "material_parse_failed",
-                    filename=str(failure.get("name") or ""),
-                    formats=tr(lang_code, formats_key),
-                )
-            )
-        if material_gate.get("content_insufficient"):
-            st.error(tr(lang_code, "material_research_content_insufficient"))
-    elif not active:
-        material_gate = {"ok": False, "failures": []}
-
     st.caption(tr(lang_code, "method_detection_explanation"))
 
     block_reason = str(preflight.get("block_reason") or "")
@@ -1914,6 +1918,85 @@ def render_workspace_controls(user):
         else:
             st.markdown(tr(lang_code, "template_help_undetermined"))
 
+    allowed_extensions = (
+        _research_material_extension_contract(method_family)
+        if method_family in METHOD_FAMILIES
+        else ()
+    )
+    uploader_help_key = (
+        f"research_materials_help_{method_family}"
+        if method_family in METHOD_FAMILIES
+        else "research_materials_help_undetermined"
+    )
+    uploaded_widget = st.file_uploader(
+        tr(lang_code, "research_materials_label"),
+        type=list(allowed_extensions),
+        accept_multiple_files=True,
+        key=f"up_research_materials_{method_family}",
+        disabled=active or not allowed_extensions,
+        help=tr(lang_code, uploader_help_key),
+    )
+    # Canonical non-widget session projection used by request freeze/preflight.
+    st.session_state["up_research_materials"] = list(uploaded_widget or [])
+
+    material_entries = _uploaded_material_entries()
+    limits_ok, material_counts, material_policy = _material_limits_ok(material_entries)
+    if material_counts["research"] > material_policy["research_material_file_limit"]:
+        st.error(
+            tr(
+                lang_code,
+                "material_research_limit_exceeded",
+                limit=material_policy["research_material_file_limit"],
+            )
+        )
+
+    material_gate = {"ok": True, "failures": []}
+    if not active and limits_ok:
+        material_gate = _material_preflight_report(method_family, material_entries)
+        expected_format = (material_gate.get("expected_extensions") or [""])[0].upper()
+        for failure in material_gate.get("failures") or []:
+            error_code = str(failure.get("error") or "")
+            if error_code.startswith("research_material_file_too_large"):
+                st.error(
+                    tr(
+                        lang_code,
+                        "material_file_size_exceeded",
+                        filename=str(failure.get("name") or ""),
+                        limit_mb=max(1, int(material_policy["research_material_file_max_bytes"]) // (1024 * 1024)),
+                    )
+                )
+            elif error_code == "template_example_not_replaced":
+                st.error(
+                    tr(
+                        lang_code,
+                        "material_template_example_not_replaced",
+                        filename=str(failure.get("name") or ""),
+                    )
+                )
+            else:
+                st.error(
+                    tr(
+                        lang_code,
+                        "material_template_parse_failed",
+                        filename=str(failure.get("name") or ""),
+                        format=expected_format,
+                    )
+                )
+        if material_gate.get("total_size_exceeded"):
+            st.error(
+                tr(
+                    lang_code,
+                    "material_total_size_exceeded",
+                    limit_mb=max(1, int(material_policy["research_material_total_max_bytes"]) // (1024 * 1024)),
+                )
+            )
+        if material_gate.get("bundle_capacity_exceeded"):
+            st.error(tr(lang_code, "material_bundle_capacity_exceeded"))
+        if material_gate.get("content_insufficient"):
+            st.error(tr(lang_code, "material_research_content_insufficient"))
+    elif not active:
+        material_gate = {"ok": False, "failures": []}
+
     product_ui = get_product_ui_projection()
     product_ui_ready = product_ui.get("mode") == "a5_product"
     selected_remaining: int | None = None

config.py

@@ -68,21 +68,28 @@ def load_config():
         "rag_chunk_size": int(os.environ.get("RAG_CHUNK_SIZE", "3000")),
         "rag_top_k": int(os.environ.get("RAG_TOP_K", "3")),
         "max_extract_chars": int(os.environ.get("MAX_EXTRACT_CHARS", "150000")),
-        # P0-R7-B: upload-role limits are policy-driven and centrally owned.
-        # app.py reads this policy; it must not embed business-count literals.
+        # P0-R7-E1: research-material-only upload policy.  File type is owned
+        # by the selected Research Design capability; app.py owns no format or
+        # count business literals.
         "research_material_upload_policy": {
             "version": os.environ.get(
                 "SMARTPAPER_RESEARCH_MATERIAL_UPLOAD_POLICY_VERSION",
-                "research-material-upload-policy-v1",
-            ).strip() or "research-material-upload-policy-v1",
-            "reference_file_limit": max(
-                1,
-                min(100, int(os.environ.get("SMARTPAPER_REFERENCE_FILE_LIMIT", "10"))),
-            ),
+                "research-material-upload-policy-v2-template-strict",
+            ).strip() or "research-material-upload-policy-v2-template-strict",
             "research_material_file_limit": max(
                 1,
                 min(100, int(os.environ.get("SMARTPAPER_RESEARCH_MATERIAL_FILE_LIMIT", "10"))),
             ),
+            # P0-R7-E1-R2: application-level byte gates are policy-driven and
+            # evaluated before any strict parser reads user content.
+            "research_material_file_max_bytes": max(
+                1024,
+                int(os.environ.get("SMARTPAPER_RESEARCH_MATERIAL_FILE_MAX_BYTES", str(2 * 1024 * 1024))),
+            ),
+            "research_material_total_max_bytes": max(
+                1024,
+                int(os.environ.get("SMARTPAPER_RESEARCH_MATERIAL_TOTAL_MAX_BYTES", str(10 * 1024 * 1024))),
+            ),
             # Preserve the existing qualitative engine's >100-character
             # user-evidence threshold without duplicating it in UI code.
             "qualitative_min_combined_chars": max(
@@ -97,7 +104,6 @@ def load_config():
                 1000,
                 int(os.environ.get("SMARTPAPER_QUALITATIVE_RESEARCH_BUNDLE_MAX_CHARS", os.environ.get("MAX_EXTRACT_CHARS", "150000"))),
             ),
-            "accepted_extensions": ["pdf", "txt", "doc", "docx", "csv", "xls", "xlsx"],
         },
         # Model requests may legitimately run longer than one Worker lease.
         # The independent Executor renews authority continuously; model-token

core/advanced_method_registry.py

@@ -34,6 +34,7 @@ _METHOD_CAPABILITIES: dict[str, dict[str, Any]] = {
         "engine_family": "qualitative_engine",
         "template_profile": "qualitative_materials",
         "material_profile": "qualitative_materials",
+        "research_material_extensions": ("txt",),
         "material_routing": {
             "structured": "data",
             "text": "interview",
@@ -51,6 +52,7 @@ _METHOD_CAPABILITIES: dict[str, dict[str, Any]] = {
         "engine_family": "quantitative_engine",
         "template_profile": "survey_quantitative_data",
         "material_profile": "survey_quantitative_data",
+        "research_material_extensions": ("csv",),
         "material_routing": {
             "structured": "data",
             "text": "reference",
@@ -342,6 +344,11 @@ def resolve_generation_preflight(
     engine_family = str(capability.get("engine_family") or "").strip()
     template_profile = str(capability.get("template_profile") or "").strip()
     material_profile = str(capability.get("material_profile") or "").strip()
+    research_material_extensions = tuple(
+        str(item or "").strip().casefold()
+        for item in (capability.get("research_material_extensions") or ())
+        if str(item or "").strip()
+    )
     material_routing = dict(capability.get("material_routing") or {})
     routing_keys = {
         "structured", "text", "document_reference", "document_research",
@@ -355,6 +362,8 @@ def resolve_generation_preflight(
         or not engine_family
         or not template_profile
         or not material_profile
+        or len(research_material_extensions) != 1
+        or research_material_extensions[0] not in {"txt", "csv"}
         or set(material_routing) != routing_keys
         or not routing_values.issubset({"reference", "data", "interview"})
     ):

core/qualitative_engine.py

@@ -664,6 +664,23 @@ Rules:
 {schema}
 """
 
+    @staticmethod
+    def _quote_has_substantive_content(value):
+        """Language-aware minimum-content gate for participant quotations.
+
+        The previous whitespace-token rule rejected ordinary Chinese sentences
+        because an entire CJK sentence can contain zero spaces.  Article Agent
+        supports Chinese/English/Russian/Kazakh, so accept either four or more
+        whitespace-delimited tokens or a substantive CJK character sequence.
+        """
+        quote = re.sub(r"\s+", " ", str(value or "")).strip()
+        if not quote:
+            return False
+        if len(quote.split()) >= 4:
+            return True
+        cjk_chars = re.findall(r"[\u3400-\u4dbf\u4e00-\u9fff]", quote)
+        return len(cjk_chars) >= 8
+
     def _qualitative_packet_validation_error(self, data, *, has_user_data, user_intent=None):
         """Return a bounded local validation reason, or an empty string when safe.
 
@@ -739,7 +756,7 @@ Rules:
                 continue
             quote = self._clean_cell(item.get("quote"), "").replace('"', "'")
             normalized_quote = re.sub(r"\s+", " ", quote).strip().casefold()
-            if not quote or len(quote.split()) < 4 or normalized_quote in seen_quotes:
+            if not self._quote_has_substantive_content(quote) or normalized_quote in seen_quotes:
                 continue
             seen_quotes.add(normalized_quote)
             retained += 1
@@ -783,12 +800,13 @@ Rules:
             else:
                 retry_prompt = base_prompt + (
                     "\n\nLOCAL VALIDATION RETRY: The previous completed response failed local "
-                    "JSON/schema/content validation. Generate a fresh complete qualitative evidence "
-                    "packet from the ORIGINAL inputs above. Do not repair, continue, quote, or rely on "
-                    "the previous response. Return exactly one complete JSON object matching the required "
-                    "schema, with no Markdown or explanatory prose. Preserve every evidence-fidelity rule "
-                    "from the original request, including verbatim quotation requirements when user material "
-                    "was supplied."
+                    f"JSON/schema/content validation with reason={last_reason}. Generate a fresh complete "
+                    "qualitative evidence packet from the ORIGINAL inputs above. Do not repair, continue, "
+                    "quote, or rely on the previous response. Return exactly one complete JSON object matching "
+                    "the required schema, with no Markdown or explanatory prose. The evidence array must contain "
+                    "the required complete evidence objects, each with non-empty theme, concept, quote, stage, "
+                    "case_label, and evidence_role fields. Preserve every evidence-fidelity rule from the original "
+                    "request, including verbatim quotation requirements when user material was supplied."
                 )
                 response = self.agent.call_deepseek(
                     system_prompt,
@@ -917,7 +935,7 @@ Rules:
             if evidence_role not in {"typical", "deviant"}:
                 evidence_role = "typical"
             normalized_quote = re.sub(r"\s+", " ", quote).strip().lower()
-            if not quote or len(quote.split()) < 4 or normalized_quote in seen_quotes:
+            if not self._quote_has_substantive_content(quote) or normalized_quote in seen_quotes:
                 continue
             seen_quotes.add(normalized_quote)
             tag = f"QUOTE-{len(clean_evidence) + 1:02d}"

i18n.py

@@ -277,21 +277,29 @@ TEXTS: dict[str, dict[str, str]] = {
         "asq_hint": "**⚠️ 核心提示**：为了让AI准确提取语录和编码树，请务必标注**受访者角色（Role）**，并采用清晰的**对话格式**。",
         "download_asq_template": "📥 一键下载 TXT 访谈模板",
         "start_asq": "🚀 启动 ASQ 模式生成",
-        "research_materials_label": "研究资料（可选）",
-        "research_materials_help": "支持 PDF、TXT、DOC、DOCX、CSV、XLS、XLSX。上传资料用于辅助论文生成。研究设计默认为质性研究，可根据需要调整为问卷量化研究。",
+        "research_materials_label": "📂 研究资料（可选）",
+        "research_materials_help": "研究资料必须使用上方推荐模板。",
+        "research_materials_help_qualitative": "质性研究仅接受 TXT，且内容必须符合上方推荐模板。",
+        "research_materials_help_survey_path": "问卷量化研究仅接受 CSV，且内容必须符合上方推荐模板。",
+        "research_materials_help_undetermined": "请先确认研究设计。",
         "drop_research_materials": "将研究资料拖到这里",
-        "limit_research_materials": "每个文件最大 1 GB · PDF, TXT, DOC, DOCX, CSV, XLS, XLSX",
+        "limit_research_materials": "每个文件最大 1 GB · 仅接受当前研究设计对应模板格式",
         "material_document_use_label": "如何使用上传的 PDF/DOC/DOCX 文件？",
         "material_use_reference": "作为参考文献资料",
         "material_use_research": "作为研究资料",
         "material_reference_limit_exceeded": "参考文献最多可选择 {limit} 个，请调整文件用途或删除多余文件。",
         "material_research_limit_exceeded": "研究资料最多可选择 {limit} 个，请调整文件用途或删除多余文件。",
         "material_parse_failed": "文件“{filename}”无法被系统正确解析。请先转换为可识别格式后重新上传（建议：{formats}）。",
+        "material_template_parse_failed": "研究资料“{filename}”无法按推荐模板正确识别。请使用上方推荐模板整理为 {format} 文件后重新上传。",
         "material_reference_convert_formats": "PDF、DOCX 或 TXT",
         "material_research_convert_formats": "PDF、DOCX、TXT、CSV 或 XLSX",
         "material_research_content_insufficient": "已选择的研究资料可解析内容不足，系统无法可靠识别。请补充内容或转换为可正确识别的格式后重新上传。",
+        "material_file_size_exceeded": "研究资料“{filename}”文件过大。单个文件不得超过 {limit_mb} MB，请精简后重新上传。",
+        "material_total_size_exceeded": "研究资料总大小超过 {limit_mb} MB，请减少文件数量或精简内容后重新上传。",
+        "material_bundle_capacity_exceeded": "研究资料内容超过当前研究设计可可靠处理的容量。请精简材料后重新上传，系统不会静默截断。",
+        "material_template_example_not_replaced": "研究资料“{filename}”仍是未修改的推荐模板示例。请用真实研究结果替换示例内容后重新上传。",
         "material_preflight_blocked": "研究资料预检未通过，请先处理上方文件问题；系统尚未创建任务或预留额度。",
-        "detected_method_label": "研究设计（可调整）",
+        "detected_method_label": "🧭 研究设计（可调整）",
         "method_detection_explanation": "默认为质性研究；生成前可根据需要调整为问卷量化研究。",
         "method_confirm_label": "调整研究设计",
         "method_confirm_placeholder": "请选择研究设计",
@@ -532,21 +540,29 @@ TEXTS: dict[str, dict[str, str]] = {
         "asq_hint": "**⚠️ Important**: Clearly label each **informant role** and use a consistent **dialogue format** so the system can accurately extract quotations and build the coding structure.",
         "download_asq_template": "📥 Download TXT Interview Template",
         "start_asq": "🚀 Start ASQ Generation",
-        "research_materials_label": "Research Materials (Optional)",
-        "research_materials_help": "Supports PDF, TXT, DOC, DOCX, CSV, XLS, and XLSX. Uploaded files support paper generation. Research Design defaults to Qualitative Research and can be adjusted to Survey-Based Quantitative Research.",
+        "research_materials_label": "📂 Research Materials (Optional)",
+        "research_materials_help": "Research materials must follow the Recommended Template above.",
+        "research_materials_help_qualitative": "Qualitative Research accepts TXT only, following the Recommended Template above.",
+        "research_materials_help_survey_path": "Survey-Based Quantitative Research accepts CSV only, following the Recommended Template above.",
+        "research_materials_help_undetermined": "Confirm the Research Design first.",
         "drop_research_materials": "Drag research materials here",
-        "limit_research_materials": "Up to 1 GB per file · PDF, TXT, DOC, DOCX, CSV, XLS, XLSX",
+        "limit_research_materials": "Up to 1 GB per file · only the template format for the current Research Design",
         "material_document_use_label": "How should SmartPaper use the uploaded PDF/DOC/DOCX files?",
         "material_use_reference": "Use as reference literature",
         "material_use_research": "Use as research materials",
         "material_reference_limit_exceeded": "You can select at most {limit} reference files. Change file roles or remove extra files.",
         "material_research_limit_exceeded": "You can select at most {limit} research-material files. Change file roles or remove extra files.",
         "material_parse_failed": "The system cannot reliably parse “{filename}”. Convert it to a supported readable format and upload it again (recommended: {formats}).",
+        "material_template_parse_failed": "Research material “{filename}” does not match the Recommended Template. Reformat it as a {format} file using the template above and upload it again.",
         "material_reference_convert_formats": "PDF, DOCX, or TXT",
         "material_research_convert_formats": "PDF, DOCX, TXT, CSV, or XLSX",
         "material_research_content_insufficient": "The selected research materials do not contain enough parseable content for reliable use. Add content or convert the files to a readable format and upload them again.",
+        "material_file_size_exceeded": "Research material “{filename}” is too large. Each file must be no larger than {limit_mb} MB. Reduce it and upload again.",
+        "material_total_size_exceeded": "The total research-material size exceeds {limit_mb} MB. Reduce the number or size of files and upload again.",
+        "material_bundle_capacity_exceeded": "The research materials exceed the reliable processing capacity for this research design. Reduce the materials and upload again; the system will not silently truncate them.",
+        "material_template_example_not_replaced": "Research material “{filename}” is still the unchanged Recommended Template example. Replace the sample values with your actual research results and upload again.",
         "material_preflight_blocked": "Research-material preflight did not pass. Resolve the file issues above first; no task or quota reservation has been created.",
-        "detected_method_label": "Research Design (Adjustable)",
+        "detected_method_label": "🧭 Research Design (Adjustable)",
         "method_detection_explanation": "Qualitative Research is selected by default. Before generation, you can adjust the Research Design to Survey-Based Quantitative Research.",
         "method_confirm_label": "Adjust Research Design",
         "method_confirm_placeholder": "Select a Research Design",
@@ -787,21 +803,29 @@ TEXTS: dict[str, dict[str, str]] = {
         "asq_hint": "**⚠️ Важно**: укажите **роль информанта** и соблюдайте единый **формат диалога**, чтобы система точно извлекала цитаты и строила структуру кодирования.",
         "download_asq_template": "📥 Скачать TXT-шаблон интервью",
         "start_asq": "🚀 Запустить формирование в режиме ASQ",
-        "research_materials_label": "Материалы исследования (необязательно)",
-        "research_materials_help": "Поддерживаются PDF, TXT, DOC, DOCX, CSV, XLS и XLSX. Загруженные файлы используются как дополнительные исследовательские или справочные материалы. По умолчанию выбрано качественное исследование; при необходимости дизайн можно изменить на количественное исследование на основе опроса.",
+        "research_materials_label": "📂 Материалы исследования (необязательно)",
+        "research_materials_help": "Материалы исследования должны соответствовать рекомендуемому шаблону выше.",
+        "research_materials_help_qualitative": "Для качественного исследования принимаются только TXT-файлы по рекомендуемому шаблону выше.",
+        "research_materials_help_survey_path": "Для количественного исследования на основе опроса принимаются только CSV-файлы по рекомендуемому шаблону выше.",
+        "research_materials_help_undetermined": "Сначала подтвердите дизайн исследования.",
         "drop_research_materials": "Перетащите исследовательские материалы сюда",
-        "limit_research_materials": "До 1 ГБ на файл · PDF, TXT, DOC, DOCX, CSV, XLS, XLSX",
+        "limit_research_materials": "До 1 ГБ на файл · только формат шаблона текущего дизайна исследования",
         "material_document_use_label": "Как использовать загруженные файлы PDF/DOC/DOCX?",
         "material_use_reference": "Как научные источники",
         "material_use_research": "Как исследовательские материалы",
         "material_reference_limit_exceeded": "Можно выбрать не более {limit} файлов в качестве научных источников. Измените назначение файлов или удалите лишние.",
         "material_research_limit_exceeded": "Можно выбрать не более {limit} файлов в качестве исследовательских материалов. Измените назначение файлов или удалите лишние.",
         "material_parse_failed": "Система не может корректно распознать файл «{filename}». Преобразуйте его в поддерживаемый читаемый формат и загрузите снова (рекомендуется: {formats}).",
+        "material_template_parse_failed": "Материал «{filename}» не соответствует рекомендуемому шаблону. Оформите его по шаблону выше как файл {format} и загрузите снова.",
         "material_reference_convert_formats": "PDF, DOCX или TXT",
         "material_research_convert_formats": "PDF, DOCX, TXT, CSV или XLSX",
         "material_research_content_insufficient": "В выбранных исследовательских материалах недостаточно корректно распознаваемого содержимого. Дополните материалы или преобразуйте файлы в читаемый формат и загрузите их снова.",
+        "material_file_size_exceeded": "Файл исследовательского материала «{filename}» слишком велик. Размер одного файла не должен превышать {limit_mb} МБ. Сократите его и загрузите снова.",
+        "material_total_size_exceeded": "Общий размер исследовательских материалов превышает {limit_mb} МБ. Уменьшите количество или размер файлов и загрузите снова.",
+        "material_bundle_capacity_exceeded": "Объём исследовательских материалов превышает надёжную ёмкость обработки для выбранного дизайна. Сократите материалы и загрузите снова; система не будет молча обрезать их.",
+        "material_template_example_not_replaced": "Материал «{filename}» всё ещё содержит неизменённый пример рекомендуемого шаблона. Замените пример своими фактическими результатами и загрузите файл снова.",
         "material_preflight_blocked": "Предварительная проверка исследовательских материалов не пройдена. Сначала устраните указанные выше проблемы; задача и резерв квоты ещё не созданы.",
-        "detected_method_label": "Дизайн исследования (можно изменить)",
+        "detected_method_label": "🧭 Дизайн исследования (можно изменить)",
         "method_detection_explanation": "По умолчанию выбрано качественное исследование. До начала генерации дизайн можно изменить на количественное исследование на основе опроса.",
         "method_confirm_label": "Изменить дизайн исследования",
         "method_confirm_placeholder": "Выберите дизайн исследования",
@@ -1042,21 +1066,29 @@ TEXTS: dict[str, dict[str, str]] = {
         "asq_hint": "**⚠️ Маңызды**: жүйе дәйексөздерді дәл шығарып, кодтау құрылымын құруы үшін **ақпарат берушінің рөлін** көрсетіп, бірізді **диалог пішімін** қолданыңыз.",
         "download_asq_template": "📥 TXT сұхбат үлгісін жүктеу",
         "start_asq": "🚀 ASQ режимінде дайындауды бастау",
-        "research_materials_label": "Зерттеу материалдары (міндетті емес)",
-        "research_materials_help": "PDF, TXT, DOC, DOCX, CSV, XLS және XLSX форматтары қолдау табады. Жүктелген файлдар қосымша зерттеу немесе анықтамалық материал ретінде пайдаланылады. Әдепкі бойынша сапалық зерттеу таңдалады; қажет болса, дизайнды сауалнамаға негізделген сандық зерттеуге өзгертуге болады.",
+        "research_materials_label": "📂 Зерттеу материалдары (міндетті емес)",
+        "research_materials_help": "Зерттеу материалдары жоғарыдағы ұсынылатын үлгіге сәйкес болуы тиіс.",
+        "research_materials_help_qualitative": "Сапалық зерттеу үшін жоғарыдағы ұсынылатын үлгіге сәйкес тек TXT файлы қабылданады.",
+        "research_materials_help_survey_path": "Сауалнамаға негізделген сандық зерттеу үшін жоғарыдағы ұсынылатын үлгіге сәйкес тек CSV файлы қабылданады.",
+        "research_materials_help_undetermined": "Алдымен зерттеу дизайнын растаңыз.",
         "drop_research_materials": "Зерттеу материалдарын осы жерге сүйреп әкеліңіз",
-        "limit_research_materials": "Әр файлға 1 ГБ дейін · PDF, TXT, DOC, DOCX, CSV, XLS, XLSX",
+        "limit_research_materials": "Әр файлға 1 ГБ дейін · тек ағымдағы зерттеу дизайнының үлгі форматы",
         "material_document_use_label": "Жүктелген PDF/DOC/DOCX файлдарын қалай пайдалану керек?",
         "material_use_reference": "Ғылыми дереккөз ретінде",
         "material_use_research": "Зерттеу материалы ретінде",
         "material_reference_limit_exceeded": "Ғылыми дереккөз ретінде ең көбі {limit} файл таңдауға болады. Файлдардың мақсатын өзгертіңіз немесе артық файлдарды жойыңыз.",
         "material_research_limit_exceeded": "Зерттеу материалы ретінде ең көбі {limit} файл таңдауға болады. Файлдардың мақсатын өзгертіңіз немесе артық файлдарды жойыңыз.",
         "material_parse_failed": "Жүйе «{filename}» файлын сенімді түрде тани алмады. Оны оқылатын қолдау көрсетілетін пішімге түрлендіріп, қайта жүктеңіз (ұсынылады: {formats}).",
+        "material_template_parse_failed": "«{filename}» зерттеу материалы ұсынылатын үлгіге сәйкес емес. Оны жоғарыдағы үлгімен {format} файлы ретінде қайта рәсімдеп, қайта жүктеңіз.",
         "material_reference_convert_formats": "PDF, DOCX немесе TXT",
         "material_research_convert_formats": "PDF, DOCX, TXT, CSV немесе XLSX",
         "material_research_content_insufficient": "Таңдалған зерттеу материалдарында сенімді пайдалануға жеткілікті танылатын мазмұн жоқ. Мазмұнды толықтырыңыз немесе файлдарды оқылатын пішімге түрлендіріп, қайта жүктеңіз.",
+        "material_file_size_exceeded": "«{filename}» зерттеу материалының файлы тым үлкен. Бір файл {limit_mb} МБ-тан аспауы тиіс. Қысқартып, қайта жүктеңіз.",
+        "material_total_size_exceeded": "Зерттеу материалдарының жалпы көлемі {limit_mb} МБ-тан асады. Файл санын немесе көлемін азайтып, қайта жүктеңіз.",
+        "material_bundle_capacity_exceeded": "Зерттеу материалдарының көлемі таңдалған дизайн үшін сенімді өңдеу шегінен асады. Материалдарды қысқартып, қайта жүктеңіз; жүйе оларды үнсіз қысқартпайды.",
+        "material_template_example_not_replaced": "«{filename}» материалы әлі де өзгертілмеген ұсынылатын үлгі мысалы болып тұр. Үлгі мәндерін нақты зерттеу нәтижелеріңізбен ауыстырып, қайта жүктеңіз.",
         "material_preflight_blocked": "Зерттеу материалдарының алдын ала тексеруі өтпеді. Алдымен жоғарыдағы файл мәселелерін түзетіңіз; тапсырма да, квота резерві де әлі жасалған жоқ.",
-        "detected_method_label": "Зерттеу дизайны (өзгертуге болады)",
+        "detected_method_label": "🧭 Зерттеу дизайны (өзгертуге болады)",
         "method_detection_explanation": "Әдепкі бойынша сапалық зерттеу таңдалады. Генерацияны бастамас бұрын зерттеу дизайнын сауалнамаға негізделген сандық зерттеуге өзгертуге болады.",
         "method_confirm_label": "Зерттеу дизайнын өзгерту",
         "method_confirm_placeholder": "Зерттеу дизайнын таңдаңыз",

utils/file_parser.py

@@ -6,9 +6,11 @@ an explicit validator that the UI can run before task creation/quota reservation
 """
 from __future__ import annotations
 
+import hashlib
 import io
 import logging
 import os
+import re
 from pathlib import Path
 from typing import BinaryIO, Iterable
 
@@ -187,6 +189,152 @@ def validate_file_parseable(file, max_chars=4000) -> dict:
         }
 
 
+
+
+_STRICT_RESEARCH_TEMPLATE_EXTENSIONS = {
+    "qualitative_materials": ".txt",
+    "survey_quantitative_data": ".csv",
+}
+
+
+def _validate_qualitative_template_text(text: str) -> None:
+    value = str(text or "").strip()
+    if not value:
+        raise FileParseError("template_content_empty")
+    metadata_lines = re.findall(r"(?m)^\s*\[[^\]\n]{2,100}\]\s*[：:]\s*\S+", value)
+    dialogue_lines = [
+        line.strip()
+        for line in value.splitlines()
+        if line.strip() and not line.lstrip().startswith("[") and re.search(r"[：:]", line)
+    ]
+    if not re.search(r"\bINT-\d+\b", value, flags=re.I):
+        raise FileParseError("template_missing_interview_id")
+    if len(metadata_lines) < 3 or len(dialogue_lines) < 2:
+        raise FileParseError("template_structure_mismatch")
+    # The downloaded template contains explicit placeholders.  If any remain,
+    # the file is not yet research material and must not enter generation.
+    if re.search(r"\[(?:在这里输入|Enter\b|Введите\b|[^\]\n]*осында енгізіңіз)", value, flags=re.I):
+        raise FileParseError("template_placeholders_not_replaced")
+
+
+def _validate_survey_template_text(text: str) -> None:
+    value = str(text or "").strip()
+    if not value:
+        raise FileParseError("template_content_empty")
+    lines = [line.strip() for line in value.splitlines() if line.strip()]
+    section_markers = [line for line in lines if line.startswith("---") and line.endswith("---")]
+    has_descriptive = any(",SD," in line and line.count(",") >= 4 for line in lines)
+    has_reliability = any("Cronbach's Alpha" in line and ",CR," in line and ",AVE" in line for line in lines)
+    has_hypothesis = any("Beta" in line and ("P-value" in line or "P-значение" in line or "P-мәні" in line) for line in lines)
+    if len(section_markers) < 3 or not (has_descriptive and has_reliability and has_hypothesis):
+        raise FileParseError("template_structure_mismatch")
+
+
+def _research_material_size_hint(file) -> int | None:
+    if isinstance(file, (str, os.PathLike, Path)):
+        try:
+            return int(Path(file).stat().st_size)
+        except OSError:
+            return None
+    value = getattr(file, "size", None)
+    try:
+        if value is not None:
+            return int(value)
+    except (TypeError, ValueError):
+        pass
+    try:
+        return int(file.getbuffer().nbytes)
+    except Exception:
+        return None
+
+
+def research_material_text_fingerprint(text: str) -> str:
+    """Stable fingerprint for detecting an unchanged downloadable template."""
+    normalized = str(text or "").replace("\r\n", "\n").replace("\r", "\n")
+    normalized = "\n".join(line.rstrip() for line in normalized.splitlines()).strip()
+    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
+
+
+def _strict_research_material_text(file, template_profile: str, *, max_bytes: int | None = None) -> tuple[str, int]:
+    profile = str(template_profile or "").strip().casefold()
+    expected_ext = _STRICT_RESEARCH_TEMPLATE_EXTENSIONS.get(profile)
+    if not expected_ext:
+        raise FileParseError("template_profile_unsupported")
+    filename = _name(file)
+    suffix = Path(filename).suffix.casefold()
+    if suffix != expected_ext:
+        raise FileParseError(f"template_extension_mismatch:expected={expected_ext}")
+    byte_limit = int(max_bytes or 0)
+    size_hint = _research_material_size_hint(file)
+    if byte_limit > 0 and size_hint is not None and size_hint > byte_limit:
+        raise FileParseError(f"research_material_file_too_large:{size_hint}:{byte_limit}")
+    try:
+        raw = _read_bytes(file)
+        if byte_limit > 0 and len(raw) > byte_limit:
+            raise FileParseError(f"research_material_file_too_large:{len(raw)}:{byte_limit}")
+        value = _decode_text_bytes(raw).strip()
+    finally:
+        if not isinstance(file, (str, os.PathLike, Path)):
+            try:
+                file.seek(0)
+            except Exception:
+                pass
+    if profile == "qualitative_materials":
+        _validate_qualitative_template_text(value)
+    elif profile == "survey_quantitative_data":
+        _validate_survey_template_text(value)
+    return value, len(raw)
+
+
+def extract_research_material_content(file, template_profile: str, max_chars=None, *, max_bytes: int | None = None) -> str:
+    """Read one strict research-material file under the selected template contract."""
+    value, _byte_size = _strict_research_material_text(file, template_profile, max_bytes=max_bytes)
+    if max_chars is None:
+        max_chars = APP_CONFIG.get("max_extract_chars", 150000)
+    max_chars = max(1, int(max_chars))
+    return value[:max_chars]
+
+
+def validate_research_material_template(
+    file,
+    template_profile: str,
+    max_chars=None,
+    *,
+    max_bytes: int | None = None,
+    disallowed_content_fingerprints: set[str] | None = None,
+) -> dict:
+    """Return an exact, UI-safe strict-template validation result.
+
+    Validation counts the full decoded text before any bundle truncation, so the
+    caller can reject oversized bundles before task creation/quota reservation.
+    """
+    filename = os.path.basename(_name(file))
+    try:
+        content, byte_size = _strict_research_material_text(file, template_profile, max_bytes=max_bytes)
+        fingerprint = research_material_text_fingerprint(content)
+        if disallowed_content_fingerprints and fingerprint in disallowed_content_fingerprints:
+            raise FileParseError("template_example_not_replaced")
+        shown = content if max_chars is None else content[: max(1, int(max_chars))]
+        return {
+            "ok": True,
+            "filename": filename,
+            "text_chars": len(content),
+            "preview_chars": len(shown),
+            "byte_size": byte_size,
+            "fingerprint": fingerprint,
+            "error": "",
+        }
+    except Exception as exc:
+        return {
+            "ok": False,
+            "filename": filename,
+            "text_chars": 0,
+            "preview_chars": 0,
+            "byte_size": int(_research_material_size_hint(file) or 0),
+            "fingerprint": "",
+            "error": str(exc)[:200],
+        }
+
 def extract_text_from_files(uploaded_files: Iterable, max_chars=None):
     if max_chars is None:
         max_chars = APP_CONFIG.get("max_extract_chars", 150000)