Blame for app.py

# -*- coding: utf-8 -*-
"""SmartPaper R3-Free UI: Streamlit is an observer, never the paper executor."""
import hashlib
import html
import json
import logging
import os
import re
import secrets
import shutil
import subprocess
import sys
import time
import textwrap
from datetime import datetime, timezone
from pathlib import Path

import streamlit as st
import streamlit.components.v1 as components

from config import APP_CONFIG
from managed_auth import (
    cancel_generation_task,
    generation_task_status,
    get_product_ui_projection,
    recover_generation_task,
    require_alipay_login,
    render_logged_in_account,
    render_payment_purchase,
    start_generation_task,
)
from i18n import (
    LANGUAGE_OPTIONS, get_output_suffix, get_research_material_template,
    get_target_flag, get_target_language, tr,
)
from core.advanced_method_registry import (
    METHOD_CAPABILITY_REGISTRY_VERSION, METHOD_FAMILIES, METHOD_RESOLVER_VERSION, default_method_family,
    method_capability, method_family_for_legacy_lane, normalize_method_family,
    reconcile_method_authority, resolve_generation_preflight,
)
from smartpaper_restore_bridge import edge_bridge
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",
    page_icon="📊",
    layout="wide",
    initial_sidebar_state="collapsed",
)


def inject_mobile_responsive_form_styles() -> None:
    """Improve narrow-screen form readability without changing desktop behavior."""
    st.markdown(
        """
        <style>
        @media (max-width: 760px) {
          /* Keep native Streamlit labels complete on narrow screens. */
          [data-testid="stTextInput"] label,
          [data-testid="stSelectbox"] label,
          [data-testid="stNumberInput"] label,
          [data-testid="stRadio"] label,
          [data-testid="stTextInput"] label p,
          [data-testid="stSelectbox"] label p,
          [data-testid="stNumberInput"] label p,
          [data-testid="stRadio"] label p {
            width: 100% !important;
            max-width: 100% !important;
            min-width: 0 !important;
            white-space: normal !important;
            overflow: visible !important;
            text-overflow: clip !important;
            overflow-wrap: anywhere !important;
            line-height: 1.35 !important;
          }

          /* Let form widgets use the full available mobile width. */
          [data-testid="stTextInput"],
          [data-testid="stSelectbox"],
          [data-testid="stNumberInput"],
          [data-testid="stRadio"] {
            width: 100% !important;
            max-width: 100% !important;
            min-width: 0 !important;
          }

          /* Preserve the native Research Design selectbox while allowing long
             selected values to wrap instead of being visually truncated. */
          [data-testid="stSelectbox"] [data-baseweb="select"],
          [data-testid="stSelectbox"] [role="combobox"] {
            width: 100% !important;
            max-width: 100% !important;
            min-width: 0 !important;
          }

          [data-testid="stSelectbox"] [data-baseweb="select"] > div {
            min-height: 3rem !important;
            height: auto !important;
          }

          [data-testid="stSelectbox"] [data-baseweb="select"] > div > div:first-child,
          [data-testid="stSelectbox"] [data-baseweb="select"] > div > div:first-child > div,
          [data-testid="stSelectbox"] [data-baseweb="select"] span,
          [data-testid="stSelectbox"] [role="combobox"] span {
            min-width: 0 !important;
            max-width: 100% !important;
            white-space: normal !important;
            overflow: visible !important;
            text-overflow: clip !important;
            overflow-wrap: anywhere !important;
            line-height: 1.3 !important;
          }

          /* BaseWeb renders the opened select menu in a portal. Keep option
             rows readable for long EN/RU/KK method names on small screens. */
          [data-baseweb="menu"] [role="option"],
          [data-baseweb="popover"] [role="option"] {
            min-height: 2.75rem !important;
            height: auto !important;
            white-space: normal !important;
            overflow-wrap: anywhere !important;
            line-height: 1.3 !important;
          }

          /* Standard / Flagship may wrap naturally rather than compressing. */
          [data-testid="stRadio"] [role="radiogroup"] {
            flex-wrap: wrap !important;
            row-gap: 0.45rem !important;
            column-gap: 1rem !important;
          }
        }
        </style>
        """,
        unsafe_allow_html=True,
    )


inject_mobile_responsive_form_styles()

# A3-R5D: default-off, benchmark-only startup probe; never blocks the UI.
try:
    from core.a3_glm_live_probe import run_startup_probe_if_enabled
    run_startup_probe_if_enabled()
except Exception as _a3_probe_exc:
    logging.error("[A3_R5D_GLM_PROBE] bootstrap_failed error_class=%s", _a3_probe_exc.__class__.__name__)

# A3-R5H: default-off Gemini Free-Tier startup probe; never blocks the UI.
try:
    from core.a3_gemini_live_probe import run_startup_probe_if_enabled as run_gemini_startup_probe_if_enabled
    run_gemini_startup_probe_if_enabled()
except Exception as _a3_gemini_probe_exc:
    logging.error("[A3_R5H_GEMINI_PROBE] bootstrap_failed error_class=%s", _a3_gemini_probe_exc.__class__.__name__)

WORKSPACE_DIR = APP_CONFIG.get("workspace_dir", ".workspace")
os.makedirs(WORKSPACE_DIR, exist_ok=True)
# The display-only preview cache must exist from the first Streamlit script run.
# Runtime preview files remain ephemeral and carry no task/execution authority.
(Path(__file__).resolve().parent / "static" / "previews").mkdir(parents=True, exist_ok=True)
os.environ["STREAMLIT_SERVER_ENABLE_CORS"] = "false"
os.environ["STREAMLIT_SERVER_ENABLE_XSRF_PROTECTION"] = "false"

WECHAT_BLOCKER_JS = """<script>try{var ua=navigator.userAgent.toLowerCase();if(ua.indexOf('micromessenger')!==-1){var parentDoc=window.parent.document;if(!parentDoc.getElementById('wx-blocker')){var overlay=parentDoc.createElement('div');overlay.id='wx-blocker';overlay.style.cssText='position:fixed;top:0;left:0;width:100vw;height:100vh;background:#1a365d;z-index:9999999;color:white;display:flex;flex-direction:column;align-items:center;justify-content:center;text-align:center;font-family:sans-serif;padding:30px;box-sizing:border-box;';overlay.innerHTML='<h2 style="font-size:26px;margin-bottom:15px;font-weight:bold;">环境受限 / Restricted</h2><p style="font-size:18px;line-height:1.6;color:#e2e8f0;max-width:100%;">为了保障最终双语 ZIP 压缩包正常下载，<b>本系统禁止在微信内直接运行</b>。<br><br>👉 <b>请点击右上角 [···] 菜单，选择“在浏览器打开” (Open in Browser)</b></p>';parentDoc.body.appendChild(overlay);}}}catch(e){}</script>"""
components.html(WECHAT_BLOCKER_JS, height=0, width=0)


def inject_file_uploader_i18n(language: str) -> None:
    """Localize uploaders by semantic label, never by visual/tab position."""
    payload = {
        "browse": tr(language, "browse_files"),
        "uploaders": {
            tr(language, "research_materials_label"): {
                "title": tr(language, "drop_research_materials"),
                "limit": tr(language, "limit_research_materials"),
            },
            # Legacy labels remain recognized for recovered historical UI state,
            # but the unified workspace renders only research_materials_label.
            tr(language, "upload_reference"): {
                "title": tr(language, "drop_docs"),
                "limit": tr(language, "limit_docs"),
            },
            tr(language, "upload_interviews"): {
                "title": tr(language, "drop_interviews"),
                "limit": tr(language, "limit_interviews"),
            },
            tr(language, "upload_stats"): {
                "title": tr(language, "drop_data"),
                "limit": tr(language, "limit_data"),
            },
        },
    }

    script = f"""
    <script>
    (() => {{
      const payload = {json.dumps(payload, ensure_ascii=False)};
      const parentDocument = window.parent.document;

      function leafTextElements(root) {{
        return Array.from(
          root.querySelectorAll("p, span, small, div")
        ).filter((element) => (
          element.children.length === 0 &&
          element.textContent.trim().length > 0
        ));
      }}

      function replaceTextSafely(element, value) {{
        if (element && element.textContent.trim() !== value) {{
          element.textContent = value;
        }}
      }}

      function semanticUploaderConfig(uploader) {{
        const label = uploader.querySelector('label');
        const labelText = (label ? label.textContent : '').trim();
        return payload.uploaders[labelText] || null;
      }}

      function applyUploaderTranslations() {{
        const uploaders = parentDocument.querySelectorAll(
          '[data-testid="stFileUploader"]'
        );

        uploaders.forEach((uploader) => {{
          const config = semanticUploaderConfig(uploader);
          if (!config) return;
          const zone = uploader.querySelector('[data-testid="stFileUploaderDropzone"]');
          if (!zone) return;

          const instructions = zone.querySelector(
            '[data-testid="stFileUploaderDropzoneInstructions"]'
          );

          if (instructions) {{
            const leaves = leafTextElements(instructions);
            replaceTextSafely(leaves[0], config.title);
            replaceTextSafely(leaves[1], config.limit);
          }}

          const button = zone.querySelector("button");
          if (button) {{
            const buttonLeaves = leafTextElements(button);
            const textTarget = buttonLeaves[buttonLeaves.length - 1];

            replaceTextSafely(textTarget, payload.browse);
            button.setAttribute("aria-label", payload.browse);
            button.setAttribute("title", payload.browse);
          }}
        }});
      }}

      [0, 100, 300, 700, 1400, 2500].forEach((delay) => {{
        window.setTimeout(applyUploaderTranslations, delay);
      }});
    }})();
    </script>
    """

    components.html(script, height=0, width=0)



def inject_wechat_browser_notice() -> None:
    """Warn immediately when SmartPaper is opened inside WeChat or WeCom."""
    script = """
    <script>
    (() => {
      const parentWindow = window.parent;
      const parentDocument = parentWindow.document;
      const userAgent = String(
        parentWindow.navigator.userAgent || navigator.userAgent || ""
      );

      const isWechat =
        /MicroMessenger|WeChat|wxwork/i.test(userAgent);

      const noticeId = "smartpaper-wechat-browser-notice";
      const existingNotice = parentDocument.getElementById(noticeId);

      if (!isWechat) {
        if (existingNotice) existingNotice.remove();
        return;
      }

      if (existingNotice) return;

      const overlay = parentDocument.createElement("div");
      overlay.id = noticeId;
      overlay.style.cssText = [
        "position:fixed",
        "inset:0",
        "z-index:2147483647",
        "display:flex",
        "align-items:center",
        "justify-content:center",
        "padding:24px",
        "background:rgba(8,12,20,.96)",
        "box-sizing:border-box"
      ].join(";");

      const card = parentDocument.createElement("div");
      card.style.cssText = [
        "width:min(620px,100%)",
        "padding:30px",
        "border:1px solid rgba(148,163,184,.35)",
        "border-radius:18px",
        "background:#111827",
        "color:#f8fafc",
        "font-family:Arial,'Microsoft YaHei','PingFang SC',sans-serif",
        "line-height:1.7",
        "box-shadow:0 20px 50px rgba(0,0,0,.38)"
      ].join(";");

      const title = parentDocument.createElement("h2");
      title.textContent = "请在浏览器中打开 / Open in Browser";
      title.style.cssText =
        "margin:0 0 14px;font-size:24px;color:#ffffff;";

      const chinese = parentDocument.createElement("p");
      chinese.textContent =
        "微信内置页面可能无法正常完成登录和文件操作。请点击右上角“…”并选择“在浏览器打开”。";
      chinese.style.cssText = "margin:0 0 12px;font-size:16px;";

      const english = parentDocument.createElement("p");
      english.textContent =
        "The WeChat in-app page may not complete sign-in or file operations correctly. Tap “…” in the top-right corner and choose “Open in Browser.”";
      english.style.cssText =
        "margin:0 0 20px;font-size:15px;color:#cbd5e1;";

      const copyButton = parentDocument.createElement("button");
      copyButton.type = "button";
      copyButton.textContent = "复制网址 / Copy link";
      copyButton.style.cssText = [
        "width:100%",
        "padding:12px 16px",
        "border:0",
        "border-radius:10px",
        "background:#1677ff",
        "color:#ffffff",
        "font-size:16px",
        "font-weight:700",
        "cursor:pointer"
      ].join(";");

      copyButton.addEventListener("click", async () => {
        const url = parentWindow.location.href;

        try {
          await parentWindow.navigator.clipboard.writeText(url);
          copyButton.textContent = "已复制 / Copied";
        } catch (_) {
          const textarea = parentDocument.createElement("textarea");
          textarea.value = url;
          textarea.style.position = "fixed";
          textarea.style.opacity = "0";
          parentDocument.body.appendChild(textarea);
          textarea.focus();
          textarea.select();
          parentDocument.execCommand("copy");
          textarea.remove();
          copyButton.textContent = "已复制 / Copied";
        }

        parentWindow.setTimeout(() => {
          copyButton.textContent = "复制网址 / Copy link";
        }, 1800);
      });

      card.appendChild(title);
      card.appendChild(chinese);
      card.appendChild(english);
      card.appendChild(copyButton);
      overlay.appendChild(card);
      parentDocument.body.appendChild(overlay);
    })();
    </script>
    """

    components.html(script, height=0, width=0)



for key, default in {
    "is_running": False,
    "zip_buffer": None,
    "zip_filename": "",
    "run_mode": "asq",
    "sp_generation_task": None,
    "sp_pending_launch": None,
    "sp_pending_resume_task_id": "",
    "sp_pending_cancel_task_id": "",
    "sp_payment_required": False,
    "sp_terminal_notice": None,
    "sp_task_recovery_initialized": False,
    "sp_last_ui_task_status": "",
    "sp_scroll_to_generation_action": False,
    "sp_completed_download_scroll_task_id": "",
    "sp_research_design_authority": "",
    "sp_method_selector_ui": "",
    "sp_research_design_ui_snapshot": "",
    "sp_research_intent_snapshot": "",
    "sp_material_document_use": "reference",  # legacy recovered-session compatibility only
    "sp_material_parse_cache": {},
}.items():
    if key not in st.session_state:
        st.session_state[key] = default


def _safe_key(value: str) -> str:
    cleaned = re.sub(r"[^A-Za-z0-9_.-]", "_", str(value or "").strip())
    return cleaned[:100] or "anonymous"


def _new_generation_idempotency_key() -> str:
    return f"gen_{int(time.time() * 1000)}_{secrets.token_urlsafe(18)}"


def _atomic_json(path: Path, value: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
    os.replace(temporary, path)


def _canonical_request_hash(value: dict) -> str:
    material = dict(value or {})
    material.pop("request_hash", None)
    encoded = json.dumps(material, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def _task_dir(task_id: str) -> Path:
    return Path(_user_workspace) / "tasks" / _safe_key(task_id)


def _local_resume_capability(task: dict) -> tuple[bool, str]:
    """Return whether this HF runtime has the local material required to resume.

    Worker/D1 remains authoritative for the remote task state. This function
    answers only the local capability question and never changes task/quota
    state.
    """
    task = dict(task or {})
    task_id = str(task.get("task_id") or "").strip()
    if not task_id:
        return False, "task_id_missing"

    task_dir = _task_dir(task_id)
    request_path = task_dir / "request.json"
    if not request_path.is_file():
        return False, "request_missing"
    try:
        request_payload = json.loads(request_path.read_text(encoding="utf-8"))
    except Exception:
        return False, "request_unreadable"
    if not isinstance(request_payload, dict):
        return False, "request_invalid"
    if str(request_payload.get("task_id") or "").strip() != task_id:
        return False, "request_task_mismatch"

    request_hash = str(request_payload.get("request_hash") or "").strip().lower()
    if not re.fullmatch(r"[a-f0-9]{64}", request_hash):
        return False, "request_hash_invalid"
    if request_hash != _canonical_request_hash(request_payload):
        return False, "request_hash_mismatch"

    task_root = task_dir.resolve()
    for field in ("reference_files", "data_files", "interview_files"):
        items = request_payload.get(field) or []
        if not isinstance(items, list):
            return False, f"{field}_invalid"
        for item in items:
            local_path = (task_dir / str(item)).resolve()
            if task_root not in local_path.parents or not local_path.is_file():
                return False, "input_missing"

    try:
        checkpoint_seq = int(task.get("checkpoint_seq") or 0)
    except (TypeError, ValueError):
        return False, "checkpoint_seq_invalid"
    if checkpoint_seq <= 0:
        return True, "ready"

    expected_stage = str(task.get("checkpoint_stage") or "").strip()
    try:
        candidates = sorted((task_dir / "checkpoints").glob(f"checkpoint_{checkpoint_seq:05d}_*.json"))
    except Exception:
        candidates = []
    for candidate in candidates:
        try:
            payload = json.loads(candidate.read_text(encoding="utf-8"))
            if not isinstance(payload, dict):
                continue
            if str(payload.get("request_hash") or "").strip().lower() != request_hash:
                continue
            if int(payload.get("checkpoint_seq") or 0) != checkpoint_seq:
                continue
            if expected_stage and str(payload.get("checkpoint_stage") or "").strip() != expected_stage:
                continue
            return True, "ready"
        except Exception:
            continue
    return False, "checkpoint_missing"


def _task_preview_path(task_id: str) -> str:
    """Return the unguessable local static preview URL for this task, if any."""
    request_path = _task_dir(task_id) / "request.json"
    if not request_path.is_file():
        return ""
    try:
        payload = json.loads(request_path.read_text(encoding="utf-8"))
        token = str(payload.get("preview_token") or "").strip()
        if not re.fullmatch(r"[A-Za-z0-9_-]{32,128}", token):
            return ""
        return f"/app/static/previews/{token}.json"
    except Exception:
        return ""


def _cleanup_task_artifacts(task_id: str) -> None:
    """Delete local stage artifacts only after authoritative cancellation."""
    task_dir = _task_dir(task_id)
    token = ""
    request_path = task_dir / "request.json"
    if request_path.is_file():
        try:
            payload = json.loads(request_path.read_text(encoding="utf-8"))
            token = str(payload.get("preview_token") or "").strip()
        except Exception:
            token = ""
    if re.fullmatch(r"[A-Za-z0-9_-]{32,128}", token):
        try:
            (Path(__file__).resolve().parent / "static" / "previews" / f"{token}.json").unlink(missing_ok=True)
        except Exception as exc:
            logging.warning("[SP_UI] cancel_preview_cleanup_error task_id=%s error=%s", task_id, type(exc).__name__)
    shutil.rmtree(task_dir, ignore_errors=True)


def _task_mode(task: dict) -> str:
    mode = str((task or {}).get("mode") or st.session_state.get("run_mode") or "asq").strip().lower()
    return "amj" if mode == "amj" else "asq"


def _task_method_family(task: dict) -> str:
    """Return frozen new-task method_family, with legacy resume fallback."""
    task_id = str((task or {}).get("task_id") or "").strip()
    if task_id:
        request_path = _task_dir(task_id) / "request.json"
        if request_path.is_file():
            try:
                payload = json.loads(request_path.read_text(encoding="utf-8"))
                family = str(payload.get("method_family") or "").strip().casefold()
                if family in METHOD_FAMILIES:
                    return family
            except Exception:
                pass
    mode = _task_mode(task)
    return method_family_for_legacy_lane(mode) or default_method_family()


def _task_output_path(task: dict) -> Path | None:
    task_id = str(task.get("task_id") or "").strip()
    filename = os.path.basename(str(task.get("output_filename") or "").strip())
    if not task_id or not filename:
        return None
    return _task_dir(task_id) / "outputs" / filename


def _save_uploaded(items, inputs_dir: Path, prefix: str) -> list[str]:
    """Persist uploads without copying the entire file into one Python bytes object."""
    saved = []
    for index, item in enumerate(items or []):
        if item is None:
            continue
        original = os.path.basename(str(getattr(item, "name", f"input_{index}")))
        safe_name = re.sub(r"[^A-Za-z0-9_.-]", "_", original)[:140] or f"input_{index}"
        filename = f"{prefix}_{index:02d}_{safe_name}"
        path = inputs_dir / filename
        try:
            item.seek(0)
        except Exception:
            pass
        if hasattr(item, "read"):
            with path.open("wb") as target:
                shutil.copyfileobj(item, target, length=1024 * 1024)
        else:
            path.write_bytes(bytes(item or b""))
        saved.append(str(Path("inputs") / filename))
        try:
            item.seek(0)
        except Exception:
            pass
    return saved


def _research_material_upload_policy() -> dict:
    policy = dict(APP_CONFIG.get("research_material_upload_policy") or {})
    research_limit = int(policy.get("research_material_file_limit") or 0)
    if research_limit <= 0:
        raise RuntimeError("research_material_upload_policy_invalid")
    return {
        **policy,
        "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)
        ),
        "quantitative_bundle_max_chars": max(
            1000, int(policy.get("quantitative_bundle_max_chars") or 20000)
        ),
        "qualitative_bundle_max_chars": max(
            1000, int(policy.get("qualitative_bundle_max_chars") or APP_CONFIG.get("max_extract_chars", 150000))
        ),
    }


def _uploaded_file_identity(item, occurrence: int = 0) -> str:
    provider_id = str(getattr(item, "file_id", "") or "").strip()
    name = os.path.basename(str(getattr(item, "name", "") or ""))
    size = str(getattr(item, "size", "") or "")
    mime = str(getattr(item, "type", "") or "")
    source = provider_id or f"{name}|{size}|{mime}|{occurrence}"
    return hashlib.sha256(source.encode("utf-8")).hexdigest()[:24]


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] = []
    for item in uploads:
        name = os.path.basename(str(getattr(item, "name", "") or ""))
        size = str(getattr(item, "size", "") or "")
        mime = str(getattr(item, "type", "") or "")
        signature = (name, size, mime)
        occurrence = occurrences.get(signature, 0)
        occurrences[signature] = occurrence + 1
        entries.append(
            {
                "item": item,
                "identity": _uploaded_file_identity(item, occurrence),
                "name": name,
                "extension": Path(name).suffix.lower(),
                "size_bytes": int(getattr(item, "size", 0) or 0),
                "role": "research",
            }
        )
    return entries


def _material_role_counts(entries: list[dict] | None = None) -> dict[str, int]:
    entries = list(entries if entries is not None else _uploaded_material_entries())
    # 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["research"] <= policy["research_material_file_limit"]
    return ok, counts, policy


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 {})
    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 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"]
        ext = str(entry.get("extension") or "").casefold().lstrip(".")
        cache_key = f"{entry['identity']}|{template_profile}|r2"
        live_cache_keys.add(cache_key)
        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": "research",
                    "error": str(result.get("error") or "parse_failed"),
                }
            )
        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)

    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
    if report["bundle_chars"] > bundle_limit:
        report["ok"] = False
        report["bundle_capacity_exceeded"] = True
    if (
        report["ok"]
        and counts["research"] > 0
        and research_bucket == "interview"
        and report["research_chars"] < policy["qualitative_min_combined_chars"]
    ):
        report["ok"] = False
        report["content_insufficient"] = True
    return report

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":
        total_budget = int(policy["quantitative_bundle_max_chars"])
    elif research_bucket == "interview":
        total_budget = int(policy["qualitative_bundle_max_chars"])
    else:
        raise RuntimeError("method_material_routing_unsupported")
    sections = []
    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()}")
    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]] = []
    for item in st.session_state.get("up_research_materials") or []:
        name = os.path.basename(str(getattr(item, "name", "") or ""))
        mime_type = str(getattr(item, "type", "") or "")
        if name or mime_type:
            out.append({"name": name, "mime_type": mime_type})
    return out


def _normalized_research_intent_text(value: object) -> str:
    """Normalize intent text so whitespace-only edits do not reset design."""
    return re.sub(r"\s+", " ", str(value or "")).strip()


def _method_intent_signature() -> str:
    """Stable Title/Keywords identity used by design reconciliation and launch."""
    payload = {
        "research_design_control": "single-authority-registry-projection-v1",
        "topic": _normalized_research_intent_text(
            st.session_state.get("sp_topic") or DEFAULT_TOPIC
        ),
        "keywords": _canonical_user_keywords(
            st.session_state.get("sp_keywords") or DEFAULT_KEYWORDS
        ),
    }
    return hashlib.sha256(
        json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest()


def _reconcile_research_design_authority() -> tuple[str, str, dict]:
    """Maintain one mutable new-task Research Design authority.

    The selector is only a UI projection. Explicit selector changes win over a
    simultaneous Title/Keywords edit; otherwise a genuine intent edit resets to
    the registry-configured default. Product tier, word count, language, upload,
    payment, and Generate reruns do not participate in this reconciliation.
    """
    default_family = default_method_family()

    authority = normalize_method_family(st.session_state.get("sp_research_design_authority"))
    if not authority:
        # One-time migration for live sessions created by the previous UI key.
        authority = (
            normalize_method_family(st.session_state.get("sp_method_selector"))
            or default_family
        )

    current_ui = (
        normalize_method_family(st.session_state.get("sp_method_selector_ui"))
        or authority
    )
    previous_ui = (
        normalize_method_family(st.session_state.get("sp_research_design_ui_snapshot"))
        or authority
    )
    current_signature = _method_intent_signature()
    previous_signature = str(st.session_state.get("sp_research_intent_snapshot") or "").strip()
    if not previous_signature:
        previous_signature = current_signature

    reconciliation = reconcile_method_authority(
        current_authority=authority,
        current_ui_family=current_ui,
        previous_ui_family=previous_ui,
        current_intent_signature=current_signature,
        previous_intent_signature=previous_signature,
    )
    resolved_family = normalize_method_family(
        reconciliation.get("method_family"), fallback_to_default=True
    )

    if reconciliation.get("ui_changed") or reconciliation.get("intent_changed"):
        # Any explicit design/intent change invalidates a not-yet-launched snapshot.
        st.session_state["sp_pending_launch"] = None

    st.session_state["sp_research_design_authority"] = resolved_family
    st.session_state["sp_method_selector_ui"] = resolved_family
    st.session_state["sp_research_design_ui_snapshot"] = resolved_family
    st.session_state["sp_research_intent_snapshot"] = current_signature
    return resolved_family, current_signature, reconciliation


def _current_method_preflight(*, confirmed_method_family: str | None = None) -> tuple[dict, str]:
    """Validate the current/frozen authority without re-selecting a design."""
    signature = _method_intent_signature()
    confirmed = normalize_method_family(confirmed_method_family)
    if not confirmed:
        confirmed = normalize_method_family(
            st.session_state.get("sp_research_design_authority"),
            fallback_to_default=True,
        )
    preflight = resolve_generation_preflight(
        str(st.session_state.get("sp_topic") or DEFAULT_TOPIC),
        _canonical_user_keywords(st.session_state.get("sp_keywords") or DEFAULT_KEYWORDS),
        # P0-R7-F1: new tasks are intentionally no-upload.  Never let stale
        # Streamlit widget/session state influence design resolution.
        materials=[],
        confirmed_method_family=confirmed,
    )
    return preflight, signature


def _clear_new_task_material_state() -> None:
    """Remove stale upload/session projections for the no-upload new-task UI.

    Historical/resumable tasks retain their already-frozen request.json inputs;
    this guard only prevents browser session residue from entering a new task.
    """
    st.session_state["up_research_materials"] = []
    st.session_state["sp_material_parse_cache"] = {}
    for family in METHOD_FAMILIES:
        st.session_state.pop(f"up_research_materials_{family}", None)

def _classify_and_save_research_materials(
    method_family: str, inputs_dir: Path
) -> tuple[list[str], list[str], list[str], str, list[dict]]:
    """Persist template-compatible research materials for the selected design.

    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:
        return [], [], [], "none", []

    report = _material_preflight_report(method_family, entries)
    if not report.get("ok"):
        raise RuntimeError("research_material_preflight_failed")

    original_research = _save_uploaded(
        [entry["item"] for entry in entries], inputs_dir, "research"
    )

    data: list[str] = []
    interviews: list[str] = []
    capability = method_capability(method_family)
    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]

    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,
    mode: str,
    lang_code: str,
    *,
    method_family: str,
    method_resolver_version: str,
) -> dict:
    task_id = str(task.get("task_id") or "").strip()
    numeric_id = (st.session_state.get("sp_user") or {}).get("numeric_id")
    if not task_id or not isinstance(numeric_id, int) or numeric_id <= 0:
        raise RuntimeError("task_identity_missing")
    task_dir = _task_dir(task_id)
    inputs_dir = task_dir / "inputs"
    inputs_dir.mkdir(parents=True, exist_ok=True)

    capability = method_capability(method_family)
    expected_lane = str(capability.get("legacy_policy_lane") or "").strip().lower()
    if not expected_lane or expected_lane != str(mode or "").strip().lower():
        raise RuntimeError("method_family_policy_lane_mismatch")

    # P0-R7-F1 no-upload contract for every NEW request.  Do not consult
    # Streamlit uploader/session state here: request freeze is the server-side
    # authority that closes hidden/stale upload paths.  Legacy frozen tasks are
    # unaffected because they already own their immutable request.json.
    refs: list[str] = []
    data: list[str] = []
    interviews: list[str] = []
    material_use_decision = "none"
    material_file_roles: list[dict] = []

    request_data = {
        "task_id": task_id,
        "user_id": numeric_id,
        "mode": mode,
        "method_family": method_family,
        "method_resolver_version": str(method_resolver_version or METHOD_RESOLVER_VERSION),
        "derived_policy_lane": expected_lane,
        "material_use_decision": material_use_decision,
        "material_file_roles": material_file_roles,
        "interface_language": lang_code,
        "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": _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.
        "preview_token": secrets.token_urlsafe(32),
        "reference_files": refs,
        "data_files": data,
        "interview_files": interviews,
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    request_data["request_hash"] = _canonical_request_hash(request_data)
    _atomic_json(task_dir / "request.json", request_data)
    return request_data

def _pid_alive(pid: int) -> bool:
    if not isinstance(pid, int) or pid <= 0:
        return False
    try:
        os.kill(pid, 0)
        return True
    except OSError:
        return False


def _pid_is_task_executor(pid: int, task_id: str, task_dir: Path) -> bool:
    """Best-effort Linux identity check before trusting/killing a launch PID.

    Hugging Face Spaces run on Linux. A stale PID can be reused, so a bare
    ``os.kill(pid, 0)`` is never enough to identify our executor.
    """
    if not _pid_alive(pid):
        return False
    try:
        raw = (Path("/proc") / str(pid) / "cmdline").read_bytes()
        command = raw.replace(b"\x00", b" ").decode("utf-8", errors="ignore")
    except Exception:
        return False
    return (
        "task_executor.py" in command
        and f"--task-id {task_id}" in command
        and str(task_dir.resolve()) in command
    )


def _launch_executor(task_id: str, *, force: bool = False) -> int:
    """Start a detached executor without making Streamlit its lifecycle owner.

    ``subprocess.Popen`` only proves that the OS accepted the child process.
    R3.2 therefore records the PID and returns immediately. The executor owns
    claim/heartbeat/checkpoint/finalization; Worker/D1 owns task authority and
    startup-lease expiry. Streamlit never waits for claim and never terminates
    an executor because a UI-side timer expired.
    """
    task_dir = _task_dir(task_id)
    request_path = task_dir / "request.json"
    if not request_path.is_file():
        raise RuntimeError("task_request_manifest_missing")
    request_data = json.loads(request_path.read_text(encoding="utf-8"))
    numeric_id = int(request_data.get("user_id") or 0)
    request_hash = str(request_data.get("request_hash") or "").strip()
    if numeric_id <= 0 or not request_hash:
        raise RuntimeError("task_request_identity_invalid")

    launch_marker_path = task_dir / "executor_launch.json"
    claim_marker_path = task_dir / "executor_claim.json"
    if launch_marker_path.is_file():
        try:
            marker = json.loads(launch_marker_path.read_text(encoding="utf-8"))
            pid = int(marker.get("pid") or 0)
            if _pid_is_task_executor(pid, task_id, task_dir) and not force:
                logging.info("[SP_UI] executor_already_spawned task_id=%s pid=%s", task_id, pid)
                return pid
        except Exception:
            pass

    # A stale local claim marker can never establish ownership for a new
    # launch/resume. Epoch fencing in Worker/D1, not process termination here,
    # prevents an older executor from overwriting a newer epoch.
    try:
        claim_marker_path.unlink(missing_ok=True)
    except TypeError:
        if claim_marker_path.exists():
            claim_marker_path.unlink()

    log_path = task_dir / "executor.log"
    task_dir.mkdir(parents=True, exist_ok=True)
    command = [
        sys.executable,
        str(Path(__file__).resolve().parent / "task_executor.py"),
        "--task-dir", str(task_dir.resolve()),
        "--task-id", str(task_id),
        "--user-id", str(numeric_id),
    ]
    executor_env = os.environ.copy()
    executor_env["SMARTPAPER_EXECUTOR_LOG_PATH"] = str(log_path.resolve())
    process = subprocess.Popen(
        command,
        cwd=str(Path(__file__).resolve().parent),
        stdin=subprocess.DEVNULL,
        # Inherit stdout/stderr so executor bootstrap/errors are visible in HF
        # Container Logs; task_executor also writes the same log to executor.log.
        close_fds=True,
        start_new_session=True,
        env=executor_env,
    )
    _atomic_json(
        launch_marker_path,
        {
            "pid": process.pid,
            "task_id": str(task_id),
            "request_hash": request_hash,
            "launched_at": datetime.now(timezone.utc).isoformat(),
        },
    )
    logging.info("[SP_UI] executor_spawned task_id=%s pid=%s", task_id, process.pid)
    return int(process.pid)


def _process_pending_cancel() -> None:
    task_id = str(st.session_state.get("sp_pending_cancel_task_id") or "").strip()
    if not task_id:
        return
    result = cancel_generation_task(task_id)
    if result.get("cancelled"):
        _cleanup_task_artifacts(task_id)
    st.session_state["sp_pending_cancel_task_id"] = ""
    st.session_state["sp_generation_task"] = dict(result.get("task") or {}) or None
    st.rerun()


def _process_pending_payment_dialog(lang_code: str) -> None:
    """Reopen purchase dialog until the initial browser bridge creates its trusted order."""
    if not bool(st.session_state.get("sp_payment_dialog_pending")):
        return
    render_payment_purchase(lang_code)


def _process_pending_resume() -> None:
    task_id = str(st.session_state.get("sp_pending_resume_task_id") or "").strip()
    if not task_id:
        return
    task = dict(st.session_state.get("sp_generation_task") or {})
    capable, reason = _local_resume_capability(task)
    if str(task.get("task_id") or "").strip() != task_id or not capable:
        logging.warning("[SP_UI] resume_unavailable task_id=%s reason=%s", task_id, reason)
        st.session_state["sp_pending_resume_task_id"] = ""
        lang_code = str(st.session_state.get("sp_lang_code") or "zh")
        st.session_state["sp_terminal_notice"] = {
            "kind": "warning",
            "message": tr(lang_code, "resume_unavailable_runtime_replaced"),
        }
        st.rerun()
    _launch_executor(task_id, force=True)
    st.session_state["sp_pending_resume_task_id"] = ""
    st.session_state["sp_scroll_to_generation_action"] = True
    st.rerun()


def _process_pending_launch(lang_code: str) -> None:
    pending = st.session_state.get("sp_pending_launch")
    if not isinstance(pending, dict):
        return

    pending_family = str(pending.get("method_family") or "").strip().casefold()
    pending_mode = str(pending.get("mode") or "").strip().lower()
    idempotency_key = str(pending.get("idempotency_key") or "").strip()
    pending_signature = str(pending.get("method_intent_signature") or "").strip()

    # Validate the frozen launch snapshot. This gate may reject stale state, but
    # it must never choose or rewrite Research Design after the user clicks Generate.
    current_authority = normalize_method_family(
        st.session_state.get("sp_research_design_authority")
    )
    preflight, current_signature = _current_method_preflight(
        confirmed_method_family=pending_family
    )
    resolved_family = str(preflight.get("method_family") or "").strip().casefold()
    resolved_mode = str(preflight.get("legacy_policy_lane") or "").strip().lower()
    capability = dict(preflight.get("capability") or {})
    projection_matches = all([
        str(pending.get("capability_registry_version") or "").strip()
            == str(capability.get("registry_version") or "").strip(),
        str(pending.get("engine_family") or "").strip()
            == str(capability.get("engine_family") or "").strip(),
        str(pending.get("template_profile") or "").strip()
            == str(capability.get("template_profile") or "").strip(),
        str(pending.get("material_profile") or "").strip()
            == str(capability.get("material_profile") or "").strip(),
    ])
    if (
        not bool(preflight.get("can_start_generation"))
        or not pending_family
        or pending_family != current_authority
        or pending_family != resolved_family
        or pending_mode != resolved_mode
        or not projection_matches
        or not pending_signature
        or pending_signature != current_signature
    ):
        st.session_state["sp_pending_launch"] = None
        reason = str(preflight.get("block_reason") or "method_preflight_changed")
        st.warning(tr(lang_code, "method_resolution_invalid"))
        return

    projection = get_product_ui_projection()
    selected_product_tier = (
        str(st.session_state.get("sp_product_tier") or "").strip().upper()
        if projection.get("mode") == "a5_product"
        else None
    )
    try:
        response = start_generation_task(
            idempotency_key=idempotency_key,
            mode=resolved_mode,
            interface_language=lang_code,
            target_words=int(st.session_state.get("sp_words") or APP_CONFIG["target_words_default"]),
            product_tier=selected_product_tier,
        )
    except Exception as exc:
        st.session_state["sp_pending_launch"] = None
        raw = str(exc or "")
        if "product_entitlement_required" in raw:
            st.session_state["sp_payment_required"] = True
            st.error(tr(lang_code, "product_entitlement_required"))
        elif "payment_required" in raw:
            st.session_state["sp_payment_required"] = True
            st.error(tr(lang_code, "quota_exhausted"))
        elif "a5_product_task_path_not_active" in raw or "a5_product_task_path_not_authorized" in raw:
            st.error(tr(lang_code, "product_generation_not_active"))
        else:
            st.error(raw)
        return
    task = dict(response.get("task") or {})
    task_id = str(task.get("task_id") or "").strip()
    if not task_id:
        st.session_state["sp_pending_launch"] = None
        st.error("task_start_missing_task_id")
        return
    st.session_state["sp_generation_task"] = task
    st.session_state["sp_last_ui_task_status"] = str(task.get("status") or "")
    st.session_state["run_mode"] = resolved_mode
    try:
        request_path = _task_dir(task_id) / "request.json"
        if not request_path.is_file():
            _persist_task_request(
                task,
                resolved_mode,
                lang_code,
                method_family=resolved_family,
                method_resolver_version=str(
                    (preflight.get("resolution") or {}).get("resolver_version")
                    or METHOD_RESOLVER_VERSION
                ),
            )
        _launch_executor(task_id)
    except Exception as exc:
        st.session_state["sp_pending_launch"] = None
        st.session_state["sp_pending_cancel_task_id"] = task_id
        st.session_state["sp_terminal_notice"] = {"kind": "error", "message": str(exc)[:800]}
        st.rerun()
    st.session_state["sp_pending_launch"] = None
    st.session_state["sp_payment_required"] = False
    st.session_state["sp_scroll_to_generation_action"] = True
    st.rerun()


inject_wechat_browser_notice()

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")
    return LANGUAGE_OPTIONS.get(choice, "zh")


authenticated_user = require_alipay_login(_current_language_code())
if not bool(st.session_state.get("sp_task_recovery_initialized")):
    recover_generation_task()
    st.session_state["sp_task_recovery_initialized"] = True
_recovered_task = st.session_state.get("sp_generation_task") or {}
_recovered_mode = str((_recovered_task or {}).get("mode") or "").strip().lower()
if _recovered_mode in {"amj", "asq"}:
    # The task record, not the first visual tab, owns the research mode after refresh.
    st.session_state["run_mode"] = _recovered_mode
_user_workspace = os.path.join(WORKSPACE_DIR, "users", _safe_key(authenticated_user.get("id")))
os.makedirs(_user_workspace, exist_ok=True)


ACTIVE_TASK_STATUSES = {"created", "running", "resumable", "cancelling"}
LIVE_TASK_STATUSES = {"created", "running", "cancelling"}

def _localized_task_status(lang_code: str, status: str) -> str:
    key = f"task_status_{str(status or '').strip().lower()}"
    value = tr(lang_code, key)
    return value if value != key else tr(lang_code, "draft_stage_processing")

def _localized_monitor_node(lang_code: str, node: str) -> str:
    """Translate stable executor progress codes without exposing engineering identifiers."""
    value = str(node or "").strip()
    exact = {
        "inputs_and_evidence": "draft_stage_inputs",
        "S2 + OpenAlex": "draft_stage_search",
        "literature_synthesis": "draft_stage_synthesis",
        "research_logic": "draft_stage_research_logic",
        "working_front_matter": "draft_stage_working_front",
        "outline_planning": "draft_stage_outline",
        "citation_planning": "draft_stage_citation",
        "english_validation": "draft_stage_english_validation",
        "final_front_matter": "draft_stage_final_front",
        "translation": "draft_stage_translation",
        "translation_complete": "translation_complete_label",
        "packaging": "draft_stage_packaging",
        "package_validation": "draft_stage_package_validation",
        "package_validated": "draft_stage_package_validated",
    }
    if value in exact:
        return tr(lang_code, exact[value])
    for prefix, key in (
        ("writing:", "draft_stage_writing_unit"),
        ("citation_audit:", "draft_stage_citation_audit_unit"),
        ("citation_recovery:", "draft_stage_citation_recovery_unit"),
        ("writing_complete:", "draft_stage_writing_complete_unit"),
        ("translation_final_gate:", "draft_stage_translation"),
    ):
        if value.startswith(prefix):
            parts = value[len(prefix):].split(":")
            if key == "draft_stage_translation":
                return tr(lang_code, key)
            if len(parts) >= 2:
                return tr(lang_code, key, current=parts[0], total=parts[1])
            return tr(lang_code, "draft_stage_processing")
    if value.startswith("translation_"):
        return tr(lang_code, "draft_stage_translation")
    if value.startswith("section_"):
        return tr(lang_code, "draft_stage_writing")
    return tr(lang_code, "draft_stage_processing")

def _markdown_safe_html(markup: str) -> str:
    """Prevent Streamlit Markdown from reclassifying indented HTML as code.

    Only indentation before actual HTML tags is removed. Text rendered inside
    preview containers keeps its own leading whitespace.
    """
    lines = []
    for line in textwrap.dedent(str(markup or "")).strip().splitlines():
        stripped = line.lstrip()
        lines.append(stripped if stripped.startswith("<") else line)
    return "\n".join(lines)


def _render_monitor_box(task: dict, lang_code: str) -> None:
    """Render the approved monitor location with DOM targets for live updates."""
    status = str(task.get("status") or "")
    node = str(task.get("current_node") or status or "")
    status_label = _localized_task_status(lang_code, status)
    node_label = _localized_monitor_node(lang_code, node)
    progress = max(0, min(100, int(task.get("progress_pct") or 0)))
    checkpoint_seq = int(task.get("checkpoint_seq") or 0)
    checkpoint_stage = str(task.get("checkpoint_stage") or "")
    ds_calls = int(task.get("ds_calls") or 0)
    s2_calls = int(task.get("s2_calls") or 0)
    oa_calls = int(task.get("oa_calls") or 0)

    progress_markup = ""
    if status in ACTIVE_TASK_STATUSES:
        progress_markup = textwrap.dedent(f"""
        <div id='sp-live-progress-wrap' style='margin:0 0 12px 0;'>
          <div id='sp-live-progress-text' style='margin-bottom:7px;'>{progress}% — {html.escape(node_label)}</div>
          <div style='width:100%;height:10px;background:rgba(128,128,128,.24);border-radius:999px;overflow:hidden;'>
            <div id='sp-live-progress-fill' style='width:{progress}%;height:100%;background:currentColor;border-radius:999px;transition:width .35s ease;'></div>
          </div>
        </div>
        """).strip()

    st.markdown(_markdown_safe_html(f"""
    <div class='log-box' id='sp-live-monitor'>
        {progress_markup}
        <div><b>{tr(lang_code, 'monitor_pipeline')}:</b> <span id='sp-live-status'>{html.escape(status_label)}</span></div>
        <div><b>{tr(lang_code, 'monitor_node')}:</b> <span id='sp-live-node'>{html.escape(node_label)}</span></div>
        <div><b>{tr(lang_code, 'monitor_log')}:</b> <span id='sp-live-checkpoint-stage'>{html.escape(tr(lang_code, 'monitor_checkpoint_saved', seq=checkpoint_seq))}</span><span id='sp-live-checkpoint-seq' style='display:none'>{checkpoint_seq}</span></div>
        <div id='sp-live-evidence-warning' style='display:none;margin:8px 0;padding:8px 10px;border-radius:6px;background:rgba(245,158,11,.13);'></div>
        <div id='sp-live-activity-feed' style='margin:8px 0 10px 0;max-height:150px;overflow:auto;line-height:1.55;'>
            <div>• {html.escape(node_label)}</div>
        </div><hr>
        <div><b>{tr(lang_code, 'monitor_compute')}:</b></div>
        <div style='display:flex;gap:8px;'>
            <span style='background: #fef3c7;padding:4px 8px;border-radius:4px;color:black;'>🧠 DeepSeek: <b id='sp-live-ds'>{ds_calls}</b></span>
            <span style='background: #dbeafe;padding:4px 8px;border-radius:4px;color:black;'>📚 S2: <b id='sp-live-s2'>{s2_calls}</b></span>
            <span style='background: #d1fae5;padding:4px 8px;border-radius:4px;color:black;'>🌐 OpenAlex: <b id='sp-live-oa'>{oa_calls}</b></span>
        </div>
    </div>
    """), unsafe_allow_html=True)


def _watch_task_status_live(task: dict) -> dict | None:
    """Start browser-only read polling for live UI state.

    DOM updates remain read-only and never participate in task liveness.
    During execution, progress and manuscript preview updates stay entirely in
    the browser; Streamlit is notified only once at terminal/resumable state.
    This prevents checkpoint-driven full-page reruns and visible flicker.
    """
    task_id = str(task.get("task_id") or "").strip()
    status = str(task.get("status") or "")
    if not task_id or status not in LIVE_TASK_STATUSES:
        return None
    api_origin = str(st.session_state.get("sp_auth_api_origin") or "").strip()
    if not api_origin:
        logging.warning("[SP_UI] live_watch_missing_api_origin task_id=%s", task_id)
        return None
    expected_request_hash = str(task.get("request_hash") or "").strip().lower()
    if not expected_request_hash:
        try:
            request_path = _task_dir(task_id) / "request.json"
            if request_path.is_file():
                expected_request_hash = str(json.loads(request_path.read_text(encoding="utf-8")).get("request_hash") or "").strip().lower()
        except Exception:
            expected_request_hash = ""
    try:
        watch_result = edge_bridge(
            action="task_watch",
            command_id=f"watch:{task_id}",
            api_origin=api_origin,
            payload={
                "task_id": task_id,
                "poll_interval_ms": 2000,
                "preview_interval_ms": 700,
                "preview_path": _task_preview_path(task_id),
                "expected_execution_epoch": int(task.get("execution_epoch") or 0),
                "expected_request_hash": expected_request_hash,
                "initial_checkpoint_seq": int(task.get("checkpoint_seq") or 0),
                "best_effort_page_exit_cancel": False,
                "draft_status_labels": {
                    "default": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_preparing"),
                    "inputs_and_evidence": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_inputs"),
                    "S2 + OpenAlex": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_search"),
                    "literature_synthesis": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_synthesis"),
                    "research_logic": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_research_logic"),
                    "outline_planning": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_outline"),
                    "citation_planning": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_citation"),
                    "writing": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_writing"),
                    "working_front_matter": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_working_front"),
                    "writing_unit": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_writing_unit"),
                    "citation_audit_unit": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_citation_audit_unit"),
                    "citation_recovery_unit": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_citation_recovery_unit"),
                    "writing_complete_unit": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_writing_complete_unit"),
                    "english_validation": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_english_validation"),
                    "final_front_matter": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_final_front"),
                    "translation": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_translation"),
                    "packaging": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_packaging"),
                    "package_validation": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_package_validation"),
                    "package_validated": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_package_validated"),
                    "processing": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_processing"),
                    "checkpoint_saved": tr(str(st.session_state.get("sp_lang_code") or "zh"), "monitor_checkpoint_saved"),
                    "status_created": tr(str(st.session_state.get("sp_lang_code") or "zh"), "task_status_created"),
                    "status_running": tr(str(st.session_state.get("sp_lang_code") or "zh"), "task_status_running"),
                    "status_resumable": tr(str(st.session_state.get("sp_lang_code") or "zh"), "task_status_resumable"),
                    "status_cancelling": tr(str(st.session_state.get("sp_lang_code") or "zh"), "task_status_cancelling"),
                    "status_completed": tr(str(st.session_state.get("sp_lang_code") or "zh"), "task_status_completed"),
                    "status_failed": tr(str(st.session_state.get("sp_lang_code") or "zh"), "task_status_failed"),
                    "status_cancelled": tr(str(st.session_state.get("sp_lang_code") or "zh"), "task_status_cancelled"),
                    "evidence_limited": tr(str(st.session_state.get("sp_lang_code") or "zh"), "evidence_limited_running"),
                    "progress": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_detail_progress"),
                    "s2": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_detail_s2"),
                    "oa": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_detail_oa"),
                    "current": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_detail_current"),
                },
                "target_status_labels": {
                    "waiting": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_waiting"),
                    "translating": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_section_running"),
                    "qa": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_section_qa"),
                    "final_gate": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_final_gate"),
                    "repairing": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_section_repairing"),
                    "repair_pending": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_section_repair_pending"),
                    "adjudicating": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_section_adjudicating"),
                    "counts": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_progress_counts"),
                    "ready": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_section_ready"),
                    "complete": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_complete_label"),
                },
            },
            key=f"sp_task_watch_{_safe_key(task_id)}",
        )
        boundary_task = watch_result.get("task") if isinstance(watch_result, dict) and watch_result.get("boundary") else None
        if isinstance(watch_result, dict) and watch_result.get("boundary"):
            navdiag = watch_result.get("ui_navdiag")
            if isinstance(navdiag, dict):
                try:
                    safe_diag = json.dumps(navdiag, ensure_ascii=False, separators=(",", ":"))[:4000]
                    logging.info("[SP_UI_NAVDIAG] task_id=%s snapshot=%s", task_id, safe_diag)
                except Exception:
                    logging.info("[SP_UI_NAVDIAG] task_id=%s snapshot_unserializable", task_id)
        if isinstance(boundary_task, dict) and str(boundary_task.get("task_id") or "") == task_id:
            st.session_state["sp_generation_task"] = dict(boundary_task)
            st.session_state["sp_last_ui_task_status"] = str(boundary_task.get("status") or "")
            return dict(boundary_task)
    except Exception as exc:
        logging.warning("[SP_UI] live_watch_error task_id=%s error=%s", task_id, type(exc).__name__)
    return None


def _load_confirmed_checkpoint(task: dict) -> dict | None:
    """Load only the checkpoint sequence already acknowledged by Worker/D1."""
    task_id = str(task.get("task_id") or "").strip()
    checkpoint_seq = int(task.get("checkpoint_seq") or 0)
    if not task_id or checkpoint_seq <= 0:
        return None
    task_dir = _task_dir(task_id)
    request_path = task_dir / "request.json"
    if not request_path.is_file():
        return None
    try:
        request_payload = json.loads(request_path.read_text(encoding="utf-8"))
        expected_hash = str(request_payload.get("request_hash") or "")
        candidates = sorted((task_dir / "checkpoints").glob(f"checkpoint_{checkpoint_seq:05d}_*.json"))
        if not candidates:
            return None
        payload = json.loads(candidates[-1].read_text(encoding="utf-8"))
        if not isinstance(payload, dict):
            return None
        if expected_hash and str(payload.get("request_hash") or "") != expected_hash:
            return None
        if int(payload.get("checkpoint_seq") or 0) != checkpoint_seq:
            return None
        return payload
    except Exception as exc:
        logging.warning("[SP_UI] checkpoint_preview_error task_id=%s error=%s", task_id, type(exc).__name__)
        return None


_PREVIEW_UNNUMBERED_HEADINGS = {
    "abstract", "keyword", "keywords", "references", "reference list",
    "摘要", "关键词", "关键字", "参考文献",
    "аннотация", "ключевое слово", "ключевые слова", "литература", "список литературы",
    "аңдатпа", "түйіндеме", "түйін сөздер", "түйінді сөздер", "кілт сөздер",
    "әдебиеттер", "пайдаланылған әдебиеттер",
}
_PREVIEW_UNNUMBERED_PREFIXES = ("appendix", "附录", "приложение", "қосымша")


def _normalize_preview_display_text(value: str) -> str:
    """Mirror the browser heading presentation for initial/terminal rerenders.

    Live updates are normalized by the browser bridge. A terminal Streamlit
    rerender used to fall back to raw Markdown, so completed pages could show
    ``#`` markers or lose the Arabic numbering users saw during generation.
    Keep one deterministic server-side equivalent for the initial/terminal DOM.
    """
    lines = str(value or "").replace("\r", "").split("\n")
    heading_rows = []
    for index, line in enumerate(lines):
        match = re.match(r"^(#{1,3})\s+(.+?)\s*$", line)
        if not match:
            continue
        title = re.sub(r"^\d+(?:\.\d+){0,2}\s+", "", match.group(2).strip())
        heading_rows.append((index, len(match.group(1)), title))
    if not heading_rows:
        return "\n".join(lines)

    def unnumbered(title: str) -> bool:
        key = str(title or "").strip().casefold()
        return key in _PREVIEW_UNNUMBERED_HEADINGS or any(key.startswith(prefix) for prefix in _PREVIEW_UNNUMBERED_PREFIXES)

    has_front_matter = any(unnumbered(title) for _index, _level, title in heading_rows)
    first_h1 = next((row for row in heading_rows if row[1] == 1), None)
    counters = [0, 0, 0]
    for index, level, title in heading_rows:
        is_known_unnumbered = unnumbered(title)
        is_title = bool(has_front_matter and first_h1 and index == first_h1[0] and not is_known_unnumbered)
        if is_known_unnumbered or is_title:
            lines[index] = title
            continue
        if level == 1:
            counters[0] += 1
            counters[1] = counters[2] = 0
            lines[index] = f"{counters[0]} {title}"
        elif level == 2:
            if counters[0] == 0:
                counters[0] = 1
            counters[1] += 1
            counters[2] = 0
            lines[index] = f"{counters[0]}.{counters[1]} {title}"
        else:
            if counters[0] == 0:
                counters[0] = 1
            if counters[1] == 0:
                counters[1] = 1
            counters[2] += 1
            lines[index] = f"{counters[0]}.{counters[1]}.{counters[2]} {title}"
    return "\n".join(lines)


def _render_checkpoint_preview(task: dict, lang_code: str) -> None:
    """Render stable bilingual DOM targets; browser mutates them in place."""
    checkpoint = _load_confirmed_checkpoint(task)
    english = str((checkpoint or {}).get("full_draft_en") or "").strip()
    target = str((checkpoint or {}).get("full_draft_translated") or "").strip()
    english_initial = _normalize_preview_display_text(english)
    english_status_initial = "" if english else tr(lang_code, "draft_preparing")
    english_status_display = "none" if english else "block"
    target_initial = _normalize_preview_display_text(target) if target else tr(lang_code, "translation_waiting")
    english_placeholder = "0" if english else "1"
    target_placeholder = "0" if target else "1"
    left, right = st.columns([1, 1], gap="medium")
    with left:
        st.markdown(tr(lang_code, "english_draft"))
        st.markdown(
            _markdown_safe_html(f"""
            <div id='sp-live-manuscript-shell' style='border:1px solid #e5e7eb;border-radius:8px;padding:14px 16px;min-height:220px;max-height:68vh;overflow:auto;background:#ffffff;color:#111827;'>
              <div id='sp-live-manuscript-status' style='white-space:pre-wrap;font-size:0.88em;color:#374151;opacity:1;margin:0 0 8px 0;display:{english_status_display};'>{html.escape(english_status_initial)}</div>
              <div id='sp-live-manuscript-preview' data-sp-placeholder='{english_placeholder}' style='white-space:pre-wrap;overflow-wrap:anywhere;margin:0;font-family:inherit;line-height:1.72;color:#111827;'>{html.escape(english_initial)}</div>
              <span id='sp-live-manuscript-cursor' aria-hidden='true' style='display:inline-block;width:7px;height:1.05em;background:#111827;margin-left:2px;vertical-align:-0.12em;animation:spBlink 1s steps(1,end) infinite;'></span>
            </div>
            """),
            unsafe_allow_html=True,
        )
    with right:
        st.markdown(tr(lang_code, "target_draft", flag=get_target_flag(lang_code)))
        st.markdown(
            _markdown_safe_html(f"""
            <div id='sp-live-translation-shell' style='border:1px solid #e5e7eb;border-radius:8px;padding:14px 16px;min-height:220px;max-height:68vh;overflow:auto;background:#ffffff;color:#111827;'>
              <div id='sp-live-translation-preview' data-sp-placeholder='{target_placeholder}' style='white-space:pre-wrap;overflow-wrap:anywhere;margin:0;font-family:inherit;line-height:1.72;color:#111827;'>{html.escape(target_initial)}</div>
              <span id='sp-live-translation-cursor' aria-hidden='true' style='display:inline-block;width:7px;height:1.05em;background:#111827;margin-left:2px;vertical-align:-0.12em;animation:spBlink 1s steps(1,end) infinite;'></span>
            </div>
            """),
            unsafe_allow_html=True,
        )
    st.markdown("<style>@keyframes spBlink { 0%,49% {opacity:1} 50%,100% {opacity:0} }</style>", unsafe_allow_html=True)


def _scroll_to_completed_download_once(task_id: str) -> None:
    """Scroll only after the task-scoped download anchor actually exists.

    Success identity is browser-session scoped and is written only after
    scrollIntoView executes. A missed early attempt therefore remains retryable
    on later Streamlit reruns.
    """
    task_id = str(task_id or "").strip()
    if not task_id:
        return
    components.html(
        f"""
        <script>
        (() => {{
          const parentWindow = window.parent;
          const parentDocument = parentWindow.document;
          const expectedTaskId = {json.dumps(task_id)};
          const storageKey = `sp-completed-download-scrolled:${{expectedTaskId}}`;
          const root = parentDocument.documentElement;
          if (root && root.dataset.spCompletedDownloadScrolled === expectedTaskId) return;
          try {{
            if (parentWindow.sessionStorage.getItem(storageKey) === '1') return;
          }} catch (_) {{}}
          let done = false;
          let observer = null;
          const locate = () => {{
            if (done) return true;
            const target = parentDocument.getElementById('sp-completed-download-anchor');
            if (!target || target.dataset.spTaskId !== expectedTaskId) return false;
            done = true;
            target.scrollIntoView({{behavior:'smooth', block:'center'}});
            if (root) root.dataset.spCompletedDownloadScrolled = expectedTaskId;
            try {{ parentWindow.sessionStorage.setItem(storageKey, '1'); }} catch (_) {{}}
            if (observer) observer.disconnect();
            return true;
          }};
          if (locate()) return;
          if (parentDocument.body && parentWindow.MutationObserver) {{
            observer = new parentWindow.MutationObserver(locate);
            observer.observe(parentDocument.body, {{childList:true, subtree:true}});
          }}
          [80, 180, 360, 700, 1200, 2000, 3200, 4800].forEach((delay) =>
            parentWindow.setTimeout(locate, delay)
          );
          parentWindow.setTimeout(() => {{ if (observer) observer.disconnect(); }}, 5200);
        }})();
        </script>
        """,
        height=0,
        width=0,
    )


def _render_task_status_once(lang_code: str) -> None:
    task = st.session_state.get("sp_generation_task") or {}
    task_id = str(task.get("task_id") or "").strip()
    if not task_id:
        return
    status = str(task.get("status") or "")

    st.session_state["sp_last_ui_task_status"] = status
    _render_monitor_box(task, lang_code)
    _render_checkpoint_preview(task, lang_code)

    if status in ACTIVE_TASK_STATUSES:
        try:
            result = generation_task_status(task_id)
            task = dict(result.get("task") or task)
            st.session_state["sp_generation_task"] = task
            status = str(task.get("status") or "")
        except Exception as exc:
            logging.warning("[SP_UI] status_poll_error task_id=%s error=%s", task_id, type(exc).__name__)

    if status in LIVE_TASK_STATUSES:
        boundary_task = _watch_task_status_live(task)
        if isinstance(boundary_task, dict):
            task = boundary_task
            status = str(task.get("status") or "")

    st.session_state["sp_last_ui_task_status"] = status

    if status == "resumable":
        resume_capable, _ = _local_resume_capability(task)
        st.warning(tr(lang_code, "checkpoint_warning" if resume_capable else "resume_unavailable_runtime_replaced"))
    elif status == "completed":
        output_path = _task_output_path(task)
        if output_path and output_path.is_file():
            st.markdown(
                f"<div id='sp-completed-download-anchor' data-sp-task-id='{html.escape(task_id, quote=True)}'></div>",
                unsafe_allow_html=True,
            )
            if str(task.get("safe_error_code") or "") == "W_EVIDENCE_COVERAGE_LIMITED":
                st.warning(tr(lang_code, "evidence_limited_completed"))
            st.success(tr(lang_code, "success"))
            st.download_button(label=tr(lang_code, "download"), data=output_path.read_bytes(), file_name=output_path.name, mime="application/zip")
            _scroll_to_completed_download_once(task_id)
        else:
            st.warning("The task completed, but the local output file is unavailable after runtime replacement.")
    elif status == "failed":
        error_message, translation_note = _localized_terminal_task_error(task, lang_code)
        st.error(error_message)
        if translation_note:
            st.warning(translation_note)
        recovery_path = _task_output_path(task)
        if (
            recovery_path
            and recovery_path.is_file()
            and recovery_path.name.startswith("SmartPaper_Recovery_")
            and recovery_path.suffix.lower() == ".zip"
        ):
            st.warning(tr(lang_code, "recovery_notice"))
            st.download_button(
                label=tr(lang_code, "recovery_download"),
                data=recovery_path.read_bytes(),
                file_name=recovery_path.name,
                mime="application/zip",
                key=f"sp_recovery_download_{task_id}",
            )
    elif status == "cancelled":
        detail = str(task.get("safe_error_detail") or "")
        if detail:
            st.warning(detail)


def _localized_terminal_task_error(task: dict, lang_code: str) -> tuple[str, str]:
    """Map stable business error codes to UI-language messages.

    Engineering details remain in task/log state; the user-facing terminal view
    does not expose a long English runtime exception in non-English interfaces.
    """
    code = str(task.get("safe_error_code") or "").strip()
    detail = str(task.get("safe_error_detail") or "").strip()
    raw = f"{code} {detail}"
    if "E_LITERATURE_INSUFFICIENT" in raw:
        return tr(lang_code, "literature_insufficient"), tr(lang_code, "upstream_translation_stopped")
    return detail or code or tr(lang_code, "node_error"), ""


def _inject_generation_action_scroll(mode: str | None = None) -> None:
    """Scroll immediately to the unified generation action without a server wait."""
    target_id = "sp-generation-action"
    components.html(
        f"""
        <script>
        (() => {{
          const parentDocument = window.parent.document;
          const targetId = {json.dumps(target_id)};
          let done = false;
          const locate = () => {{
            if (done) return;
            const target = parentDocument.getElementById(targetId);
            if (!target) return;
            done = true;
            target.scrollIntoView({{behavior:'auto', block:'center'}});
          }};
          [0, 40, 100, 220, 450, 800].forEach((delay) => window.parent.setTimeout(locate, delay));
        }})();
        </script>
        """,
        height=0,
        width=0,
    )


def _scroll_to_generation_action_once() -> None:
    """One-time post-launch/resume scroll to the monitor/live manuscript."""
    if not bool(st.session_state.get("sp_scroll_to_generation_action")):
        return
    st.session_state["sp_scroll_to_generation_action"] = False
    fallback_id = "sp-generation-action"
    components.html(
        f"""
        <script>
        (() => {{
          const parentDocument = window.parent.document;
          const fallbackId = {json.dumps(fallback_id)};
          let done = false;
          const locate = (allowFallback = false) => {{
            if (done) return;
            const target =
              parentDocument.getElementById('sp-task-monitor-anchor') ||
              parentDocument.getElementById('sp-live-manuscript-shell') ||
              (allowFallback ? parentDocument.getElementById(fallbackId) : null);
            if (!target) return;
            done = true;
            target.scrollIntoView({{behavior:'auto', block:'center'}});
          }};
          [0, 60, 140, 300, 650, 1000, 1600].forEach((delay) =>
            window.parent.setTimeout(() => locate(false), delay)
          );
          window.parent.setTimeout(() => locate(true), 2200);
        }})();
        </script>
        """,
        height=0,
        width=0,
    )


def render_workspace_controls(user):
    top_toolbar = st.container()
    with top_toolbar:
        language_col, account_col = st.columns([5.4, 4.6])
        with language_col:
            lang_choice = st.radio(
                "Language", list(LANGUAGE_OPTIONS.keys()), index=0, horizontal=True,
                label_visibility="collapsed", key="interface_language_choice",
            )
        lang_code = LANGUAGE_OPTIONS[lang_choice]
        st.session_state["sp_lang_code"] = lang_code
        st.session_state["sp_target_lang"] = get_target_language(lang_code)
        st.session_state["sp_output_suffix"] = get_output_suffix(lang_code)
        st.session_state["sp_target_flag"] = get_target_flag(lang_code)
        with account_col:
            render_logged_in_account(lang_code, user)

    task = st.session_state.get("sp_generation_task") or {}
    task_status = str(task.get("status") or "")
    task_mode = str(task.get("mode") or "").strip().lower()
    if task_mode in {"amj", "asq"}:
        st.session_state["run_mode"] = task_mode
    active = task_status in ACTIVE_TASK_STATUSES
    st.session_state["is_running"] = active

    st.markdown(f"<div class='main-title'><h2>{tr(lang_code, 'title')}</h2></div>", unsafe_allow_html=True)


    if task_status == "resumable":
        resume_capable, resume_reason = _local_resume_capability(task)
        if resume_capable:
            st.warning(tr(lang_code, "checkpoint_warning"))
        else:
            logging.warning(
                "[SP_UI] local_resume_capability_missing task_id=%s reason=%s",
                str(task.get("task_id") or ""), resume_reason,
            )
            st.warning(tr(lang_code, "resume_unavailable_runtime_replaced"))
        colA, colB = st.columns([1, 1])
        if colA.button(tr(lang_code, "resume"), key="resume_task", disabled=not resume_capable):
            st.session_state["sp_pending_resume_task_id"] = str(task.get("task_id") or "")
            st.rerun()
        if colB.button(tr(lang_code, "clear"), key="clear_task"):
            st.session_state["sp_pending_cancel_task_id"] = str(task.get("task_id") or "")
            st.rerun()
        st.markdown("---")

    st.text_input(
        tr(lang_code, "topic_label"),
        value=DEFAULT_TOPIC,
        disabled=active,
        key="sp_topic",
    )
    st.text_input(
        tr(lang_code, "keywords_label"),
        value=DEFAULT_KEYWORDS,
        disabled=active,
        key="sp_keywords",
    )

    # P0-R7-F1 simplified workspace: Paper Title -> Keywords -> Research Design
    # -> Standard/Flagship -> Target Word Count -> Generate.  New tasks accept
    # no uploaded literature or empirical material.  Historical task inputs
    # remain frozen in their existing request.json and are not rewritten here.
    if active:
        active_family = _task_method_family(task)
        active_capability = method_capability(active_family)
        preflight = {
            "can_start_generation": True,
            "block_reason": "",
            "method_family": active_family,
            "legacy_policy_lane": str(active_capability.get("legacy_policy_lane") or _task_mode(task)),
            "capability": active_capability,
            "resolution": {"confidence": "TASK_FROZEN", "resolver_version": METHOD_RESOLVER_VERSION},
        }
        method_signature = ""
    else:
        _clear_new_task_material_state()
        # One business authority owns the new-task research design. The widget is
        # only a UI projection; workflow/engine/template come from the registry.
        authority_family, method_signature, _reconciliation = (
            _reconcile_research_design_authority()
        )
        st.selectbox(
            tr(lang_code, "detected_method_label"),
            list(METHOD_FAMILIES),
            format_func=lambda family: tr(lang_code, f"method_{family}"),
            key="sp_method_selector_ui",
            disabled=active,
        )
        preflight, method_signature = _current_method_preflight(
            confirmed_method_family=authority_family
        )

    method_family = str(preflight.get("method_family") or "undetermined").strip().casefold()
    method_label = (
        tr(lang_code, f"method_{method_family}")
        if method_family in METHOD_FAMILIES
        else tr(lang_code, "method_undetermined")
    )
    if active:
        st.markdown(f"**{tr(lang_code, 'detected_method_label')}:** {method_label}")

    st.caption(tr(lang_code, "method_detection_explanation"))

    block_reason = str(preflight.get("block_reason") or "")
    if block_reason:
        st.warning(tr(lang_code, "method_resolution_invalid"))
    elif not active:
        st.info(tr(lang_code, "method_supported", method=method_label))

    product_ui = get_product_ui_projection()
    product_ui_ready = product_ui.get("mode") == "a5_product"
    selected_remaining: int | None = None
    if product_ui_ready:
        products = product_ui.get("products") or {}
        tier_options = [tier for tier in ("STANDARD", "FLAGSHIP") if tier in products]
        if len(tier_options) != 2:
            st.error(tr(lang_code, "product_generation_not_active"))
            st.stop()
        selected_tier = str(st.session_state.get("sp_product_tier") or "STANDARD").strip().upper()
        if selected_tier not in tier_options:
            selected_tier = "STANDARD"
            st.session_state["sp_product_tier"] = selected_tier
        selected_tier = st.radio(
            tr(lang_code, "product_tier_label"),
            tier_options,
            index=tier_options.index(selected_tier),
            format_func=lambda tier: tr(lang_code, f"product_tier_{tier.lower()}"),
            horizontal=True,
            disabled=active,
            key="sp_product_tier",
        )
        selected_product = products[selected_tier]
        entitlement = product_ui.get("entitlement") or {}
        selected_remaining = int(
            entitlement.get("standard_remaining" if selected_tier == "STANDARD" else "flagship_remaining") or 0
        )
        word_min = int(selected_product["min_body_words"])
        word_max = int(selected_product["max_body_words"])
        word_label = tr(
            lang_code,
            "words_label_product",
            min_words=word_min,
            max_words=word_max,
        )
    else:
        st.session_state.pop("sp_product_tier", None)
        word_min = int(APP_CONFIG["target_words_min"])
        word_max = int(APP_CONFIG["target_words_max"])
        word_label = tr(
            lang_code,
            "words_label",
            min_words=word_min,
            max_words=word_max,
        )
        st.info(tr(lang_code, "product_data_loading"))

    current_words = int(st.session_state.get("sp_words") or APP_CONFIG["target_words_default"])
    bounded_words = min(word_max, max(word_min, current_words))
    if st.session_state.get("sp_words") != bounded_words:
        st.session_state["sp_words"] = bounded_words
    st.number_input(
        word_label,
        min_value=word_min,
        max_value=word_max,
        value=bounded_words,
        step=1000,
        disabled=active,
        key="sp_words",
    )

    st.markdown("<div id='sp-generation-action'></div>", unsafe_allow_html=True)
    if active:
        if st.button(tr(lang_code, "cancel_generation"), key="cancel_generation_unified"):
            st.session_state["sp_pending_cancel_task_id"] = str(task.get("task_id") or "")
            st.rerun()
    else:
        entitlement_exhausted = bool(product_ui_ready and selected_remaining is not None and selected_remaining <= 0)
        if entitlement_exhausted:
            st.warning(tr(lang_code, "product_entitlement_required"))
            if st.button(f"💳 {tr(lang_code, 'payment_open')}", key="btn_open_payment_for_selected_tier"):
                st.session_state["sp_payment_dialog_pending"] = True
                st.rerun()
        can_start = bool(
            preflight.get("can_start_generation")
            and product_ui_ready
            and selected_remaining is not None
            and selected_remaining > 0
        )
        if st.button(
            tr(lang_code, "start_generation_unified"),
            disabled=not can_start,
            key="btn_generation_unified",
        ):
            # Freeze the already-reconciled authority. This is validation only;
            # Generate must never infer or reset Research Design.
            launch_family = normalize_method_family(
                st.session_state.get("sp_research_design_authority")
            )
            click_preflight, click_signature = _current_method_preflight(
                confirmed_method_family=launch_family
            )
            if (
                not launch_family
                or not bool(click_preflight.get("can_start_generation"))
                or str(click_preflight.get("method_family") or "").strip().casefold() != launch_family
            ):
                st.warning(tr(lang_code, "method_resolution_invalid"))
            else:
                resolved_mode = str(click_preflight.get("legacy_policy_lane") or "").strip().lower()
                capability = dict(click_preflight.get("capability") or {})
                st.session_state["run_mode"] = resolved_mode
                st.session_state["sp_pending_launch"] = {
                    "mode": resolved_mode,
                    "method_family": launch_family,
                    "method_resolver_version": str(
                        (click_preflight.get("resolution") or {}).get("resolver_version")
                        or METHOD_RESOLVER_VERSION
                    ),
                    "capability_registry_version": str(
                        capability.get("registry_version") or METHOD_CAPABILITY_REGISTRY_VERSION
                    ),
                    "engine_family": str(capability.get("engine_family") or ""),
                    "template_profile": str(capability.get("template_profile") or ""),
                    "material_profile": str(capability.get("material_profile") or ""),
                    "method_intent_signature": click_signature,
                    "idempotency_key": _new_generation_idempotency_key(),
                }
                st.session_state["sp_scroll_to_generation_action"] = True
                st.info(tr(lang_code, "pipeline_start"))
                _inject_generation_action_scroll()



render_workspace_controls(authenticated_user)
lang_code = str(st.session_state.get("sp_lang_code") or _current_language_code())

_notice = st.session_state.pop("sp_terminal_notice", None)
if isinstance(_notice, dict) and _notice.get("message"):
    (st.error if _notice.get("kind") == "error" else st.warning)(str(_notice.get("message")))

_process_pending_cancel()
_process_pending_resume()
_process_pending_launch(lang_code)
_process_pending_payment_dialog(lang_code)

if st.session_state.get("sp_payment_required"):
    st.warning(tr(lang_code, "product_entitlement_required"))
    if st.button(f"💳 {tr(lang_code, 'payment_open')}", key="btn_open_payment_after_rejection"):
        st.session_state["sp_payment_dialog_pending"] = True
        st.rerun()

# R3.2.4 keeps the approved task-monitor location. Active progress is updated
# inside the browser by read-only edge polling; Streamlit reruns only on a
# user action or a terminal/resumable boundary, so there is no periodic flicker.
_task_for_monitor = st.session_state.get("sp_generation_task") or {}
_task_for_monitor_id = str(_task_for_monitor.get("task_id") or "").strip()
if _task_for_monitor_id:
    st.markdown("<div id='sp-task-monitor-anchor'></div>", unsafe_allow_html=True)
    # Install the one-time navigation script before task-status bridge calls.
    # If the first bridge handshake yields via st.stop(), the browser still
    # receives the anchor and can position the user at visible task progress.
    _scroll_to_generation_action_once()
    _render_task_status_once(lang_code)