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
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,
    recover_generation_task,
    require_alipay_login,
    render_logged_in_account,
    render_payment_purchase,
    start_generation_task,
)
from i18n import (
    LANGUAGE_OPTIONS, get_amj_csv_template, get_asq_txt_template,
    get_output_suffix, get_target_flag, get_target_language, tr,
)
from smartpaper_restore_bridge import edge_bridge

st.set_page_config(
    page_title="Article Agent",
    page_icon="📊",
    layout="wide",
    initial_sidebar_state="collapsed",
)

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 Streamlit uploaders without modifying their React structure."""
    payload = {
        "browse": tr(language, "browse_files"),
        "uploaders": [
            {
                "title": tr(language, "drop_docs"),
                "limit": tr(language, "limit_docs"),
            },
            {
                "title": tr(language, "drop_data"),
                "limit": tr(language, "limit_data"),
            },
            {
                "title": tr(language, "drop_docs"),
                "limit": tr(language, "limit_docs"),
            },
            {
                "title": tr(language, "drop_interviews"),
                "limit": tr(language, "limit_interviews"),
            },
        ],
    }

    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 applyUploaderTranslations() {{
        const zones = parentDocument.querySelectorAll(
          '[data-testid="stFileUploaderDropzone"]'
        );

        zones.forEach((zone, index) => {{
          const config = payload.uploaders[index];
          if (!config) 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": "amj",
    "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,
}.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 _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 "amj").strip().lower()
    return "asq" if mode == "asq" else "amj"


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]:
    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
        data = item.getvalue() if hasattr(item, "getvalue") else item.read()
        path.write_bytes(bytes(data or b""))
        saved.append(str(Path("inputs") / filename))
        try:
            item.seek(0)
        except Exception:
            pass
    return saved


def _persist_task_request(task: dict, mode: str, lang_code: 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)

    if mode == "amj":
        refs = _save_uploaded(st.session_state.get("up_amj_ref") or [], inputs_dir, "ref")
        data = _save_uploaded([st.session_state.get("up_amj_data")], inputs_dir, "data")
        interviews = []
    else:
        refs = _save_uploaded(st.session_state.get("up_asq_ref") or [], inputs_dir, "ref")
        data = []
        interviews = _save_uploaded([st.session_state.get("up_asq_data")], inputs_dir, "interview")

    request_data = {
        "task_id": task_id,
        "user_id": numeric_id,
        "mode": mode,
        "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": str(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_resume() -> None:
    task_id = str(st.session_state.get("sp_pending_resume_task_id") or "").strip()
    if not task_id:
        return
    if not (_task_dir(task_id) / "request.json").is_file():
        st.session_state["sp_pending_resume_task_id"] = ""
        st.session_state["sp_pending_cancel_task_id"] = task_id
        st.session_state["sp_terminal_notice"] = {
            "kind": "error",
            "message": "Local checkpoint/request files are unavailable after runtime replacement; the reserved quota will be released.",
        }
        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
    mode = str(pending.get("mode") or "").strip().lower()
    idempotency_key = str(pending.get("idempotency_key") or "").strip()
    try:
        response = start_generation_task(
            idempotency_key=idempotency_key,
            mode=mode,
            interface_language=lang_code,
            target_words=int(st.session_state.get("sp_words") or APP_CONFIG["target_words_default"]),
        )
    except Exception as exc:
        st.session_state["sp_pending_launch"] = None
        raw = str(exc or "")
        if "payment_required" in raw:
            st.session_state["sp_payment_required"] = True
            st.error(tr(lang_code, "quota_exhausted"))
            render_payment_purchase(lang_code)
        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 "")
    try:
        request_path = _task_dir(task_id) / "request.json"
        if not request_path.is_file():
            _persist_task_request(task, mode, lang_code)
        _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: Evaluating a Chatbot-based Recommendation Engine for Personalized and Context-aware Education"
DEFAULT_KEYWORDS = "Ubiquitous computing; Context awareness; Deep learning; Personalized educational; Educational technologies; HCI; Recommendation system; LLMs; Generative AI; ChatGPT; Prompting engineering; Few-shot prompting"

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
_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 _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 "")
    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 = f"""
        <div id='sp-live-progress-wrap' style='margin:0 0 12px 0;'>
          <div id='sp-live-progress-text' style='margin-bottom:7px;'>{progress}% — {node}</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>
        """

    st.markdown(f"""
    {progress_markup}
    <div class='log-box' id='sp-live-monitor'>
        <div><b>{tr(lang_code, 'monitor_pipeline')}:</b> <span id='sp-live-status'>{status}</span></div>
        <div><b>{tr(lang_code, 'monitor_node')}:</b> <span id='sp-live-node'>{node}</span></div>
        <div><b>{tr(lang_code, 'monitor_log')}:</b> checkpoint <span id='sp-live-checkpoint-seq'>{checkpoint_seq}</span> · <span id='sp-live-checkpoint-stage'>{checkpoint_stage}</span></div>
        <div id='sp-live-activity-feed' style='margin:8px 0 10px 0;max-height:150px;overflow:auto;line-height:1.55;'>
            <div>• {node}</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
    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),
                "initial_checkpoint_seq": int(task.get("checkpoint_seq") or 0),
                "best_effort_page_exit_cancel": True,
                "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"),
                    "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"),
                    "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(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


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 = english if english else tr(lang_code, "draft_preparing")
    target_initial = 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(
            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;'>
              <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(
            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;'>
              <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 _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 "")
    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
    _render_monitor_box(task, lang_code)
    _render_checkpoint_preview(task, lang_code)

    if status == "resumable":
        st.warning(tr(lang_code, "checkpoint_warning"))
    elif status == "completed":
        output_path = _task_output_path(task)
        if output_path and output_path.is_file():
            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")
        else:
            st.warning("The task completed, but the local output file is unavailable after runtime replacement.")
    elif status == "failed":
        st.error(str(task.get("safe_error_detail") or task.get("safe_error_code") or tr(lang_code, "node_error")))
    elif status == "cancelled":
        detail = str(task.get("safe_error_detail") or "")
        if detail:
            st.warning(detail)


def _inject_generation_action_scroll(mode: str) -> None:
    """Scroll immediately to the active generation action without a server wait."""
    target_id = "sp-generation-action-asq" if str(mode or "").lower() == "asq" else "sp-generation-action-amj"
    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 live manuscript, never per checkpoint."""
    if not bool(st.session_state.get("sp_scroll_to_generation_action")):
        return
    st.session_state["sp_scroll_to_generation_action"] = False
    mode = str(st.session_state.get("run_mode") or "amj")
    fallback_id = "sp-generation-action-asq" if mode.lower() == "asq" else "sp-generation-action-amj"
    components.html(
        f"""
        <script>
        (() => {{
          const parentDocument = window.parent.document;
          const fallbackId = {json.dumps(fallback_id)};
          let done = false;
          const locate = () => {{
            if (done) return;
            const target = parentDocument.getElementById('sp-live-manuscript-shell') || parentDocument.getElementById(fallbackId);
            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 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 "")
    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":
        st.warning(tr(lang_code, "checkpoint_warning"))
        colA, colB = st.columns([1, 1])
        if colA.button(tr(lang_code, "resume"), key="resume_task"):
            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")
    st.number_input(
        tr(lang_code, "words_label"),
        min_value=int(APP_CONFIG["target_words_min"]),
        max_value=int(APP_CONFIG["target_words_max"]),
        value=int(APP_CONFIG["target_words_default"]),
        step=1000,
        disabled=active,
        key="sp_words",
    )

    tab_amj, tab_asq = st.tabs([tr(lang_code, "tab_amj"), tr(lang_code, "tab_asq")])
    with tab_amj:
        st.markdown(tr(lang_code, "amj_config"))
        st.info(tr(lang_code, "amj_info"))
        st.file_uploader(tr(lang_code, "upload_reference"), type=["pdf", "doc", "docx"], accept_multiple_files=True, key="up_amj_ref", disabled=active)
        st.file_uploader(tr(lang_code, "upload_stats"), type=["csv", "xlsx", "xls"], key="up_amj_data", disabled=active)
        with st.expander(tr(lang_code, "amj_expander")):
            st.markdown(tr(lang_code, "amj_hint"))
            template = get_amj_csv_template(lang_code)
            st.code(template, language="csv")
            st.download_button(label=tr(lang_code, "download_amj_template"), data=template, file_name="AMJ_Data_Template.csv", mime="text/csv", use_container_width=True)
        st.markdown("<div id='sp-generation-action-amj'></div>", unsafe_allow_html=True)
        if active and _task_mode(task) == "amj":
            if st.button(tr(lang_code, "cancel_generation"), key="cancel_amj"):
                st.session_state["sp_pending_cancel_task_id"] = str(task.get("task_id") or "")
                st.rerun()
        elif st.button(tr(lang_code, "start_amj"), disabled=active, key="btn_amj"):
            st.session_state["run_mode"] = "amj"
            st.session_state["sp_pending_launch"] = {"mode": "amj", "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("amj")

    with tab_asq:
        st.markdown(tr(lang_code, "asq_config"))
        st.info(tr(lang_code, "asq_info"))
        st.file_uploader(tr(lang_code, "upload_reference"), type=["pdf", "doc", "docx"], accept_multiple_files=True, key="up_asq_ref", disabled=active)
        st.file_uploader(tr(lang_code, "upload_interviews"), type=["txt", "pdf", "doc", "docx"], key="up_asq_data", disabled=active)
        with st.expander(tr(lang_code, "asq_expander")):
            st.markdown(tr(lang_code, "asq_hint"))
            template = get_asq_txt_template(lang_code)
            st.code(template, language="text")
            st.download_button(label=tr(lang_code, "download_asq_template"), data=template, file_name="ASQ_Interview_Template.txt", mime="text/plain", use_container_width=True)
        st.markdown("<div id='sp-generation-action-asq'></div>", unsafe_allow_html=True)
        if active and _task_mode(task) == "asq":
            if st.button(tr(lang_code, "cancel_generation"), key="cancel_asq"):
                st.session_state["sp_pending_cancel_task_id"] = str(task.get("task_id") or "")
                st.rerun()
        elif st.button(tr(lang_code, "start_asq"), disabled=active, key="btn_asq"):
            st.session_state["run_mode"] = "asq"
            st.session_state["sp_pending_launch"] = {"mode": "asq", "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("asq")

    inject_file_uploader_i18n(lang_code)


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)

if st.session_state.get("sp_payment_required"):
    render_payment_purchase(lang_code)

# 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 {}
if str(_task_for_monitor.get("task_id") or "").strip():
    st.markdown("<div id='sp-task-monitor-anchor'></div>", unsafe_allow_html=True)
    _render_task_status_once(lang_code)
    _scroll_to_generation_action_once()