Deploy SmartPaper v1.1 R3-Free

app.py

@@ -1,132 +1,47 @@
 # -*- coding: utf-8 -*-
-"""Smart Draft - Dual Track (AMJ Empirical vs ASQ Qualitative)"""
-from core.meta_extractor import extract_golden_metadata
-from core.rag_engine import initialize_rag
-from core.qualitative_engine import get_qualitative_engine
-from core.quantitative_engine import get_quantitative_engine
-from core.production_guard import build_validated_package, replace_abstract
-import streamlit as st
-
-st.set_page_config(
-    page_title="Article Agent",
-    page_icon="📊",
-    layout="wide",
-    initial_sidebar_state="collapsed",
-)
-
-import warnings
-warnings.filterwarnings("ignore", message="coroutine 'expire_cache' was never awaited")
-
-import os
-import streamlit.components.v1 as components
-import time
+"""SmartPaper R3-Free UI: Streamlit is an observer, never the paper executor."""
+import hashlib
+import json
 import logging
-import threading
+import os
 import re
-import math
-import json
 import secrets
-from datetime import datetime, timezone, timedelta
-from concurrent.futures import ThreadPoolExecutor
+import shutil
+import signal
+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, AMJ_PROMPTS, ASQ_PROMPTS
-from utils.file_parser import extract_text_from_files
-from core.pipeline_agent import SSciPipelineAgent
+from config import APP_CONFIG
 from managed_auth import (
+    cancel_generation_task,
+    generation_task_status,
     require_alipay_login,
     render_logged_in_account,
-    start_generation_task,
-    heartbeat_generation_task,
-    finish_generation_task,
     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 i18n import LANGUAGE_OPTIONS, get_amj_csv_template, get_asq_txt_template, get_output_suffix, get_target_flag, get_target_language, tr
-
-os.environ["STREAMLIT_SERVER_ENABLE_CORS"] = "false"
-os.environ["STREAMLIT_SERVER_ENABLE_XSRF_PROTECTION"] = "false"
 
-# Streamlit executes the app on its ScriptRunner thread rather than Python's
-# process main thread. Capture that thread so background worker threads cannot
-# emit browser/D1 heartbeats on behalf of work they do not own.
-STREAMLIT_SCRIPT_THREAD_ID = threading.get_ident()
+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)
-LEGACY_CHECKPOINT_FILE = os.path.join(WORKSPACE_DIR, "snapshot.json")
-CHECKPOINT_FILE = LEGACY_CHECKPOINT_FILE
-
-META_SECTIONS_BLACKLIST = set(APP_CONFIG["blacklists"]["meta_sections"])
-
-
-def update_heartbeat():
-    """Refresh D1 only while the Python generation process is actually active."""
-    if threading.get_ident() != STREAMLIT_SCRIPT_THREAD_ID:
-        return
-    now = time.time()
-    task = st.session_state.get("sp_generation_task") or {}
-    task_id = str(task.get("task_id") or "").strip()
-    interval = max(3, int(APP_CONFIG.get("task_heartbeat_interval_seconds", 5)))
-    last_remote = float(st.session_state.get("sp_last_task_heartbeat") or 0.0)
-    if not task_id or now - last_remote < interval:
-        return
-    try:
-        heartbeat_generation_task(
-            task_id,
-            current_node=str(st.session_state.get("sp_current_node") or "")[:240],
-        )
-        st.session_state["sp_last_task_heartbeat"] = now
-    except Exception as exc:
-        logging.warning(
-            "[SP_TASK] heartbeat_enqueue_failed task_id=%s error=%s",
-            task_id,
-            type(exc).__name__,
-        )
-
-
-def is_system_busy():
-    """D1 is the sole concurrency authority; no duplicate Hugging Face file lock."""
-    return False
-
-
-def clear_heartbeat():
-    """Backward-compatible no-op after removal of the duplicate local lock."""
-    return None
-
-
-def stream_with_heartbeat(generator):
-    """Refresh the D1 lease while streaming model output."""
-    last_beat = time.time()
-    for chunk in generator:
-        now = time.time()
-        if now - last_beat > 5:
-            update_heartbeat()
-            last_beat = now
-        yield chunk
-
-
-def clean_trailing_references(text, keywords=("reference", "references", "reference list", "参考文献")):
-    lines = text.split('\n')
-    for i in range(len(lines)-1, max(-1, len(lines)-100), -1):
-        if any(re.match(rf'^##?\s*{kw}', lines[i], re.IGNORECASE) for kw in keywords):
-            return '\n'.join(lines[:i]).strip()
-    return text.strip()
-
-def clean_markdown_headings(text):
-    return re.sub(r'(?m)^#+\s+(.*)$', r'**\1**', text)
-
-def _read_runtime_secret(name):
-    # Read a Hugging Face/Streamlit secret and remove copied whitespace.
-    value = os.environ.get(name, "")
-    if not value:
-        try:
-            value = st.secrets.get(name, "")
-        except Exception:
-            value = ""
-    return str(value or "").strip()
-
-
-global_ds_key = _read_runtime_secret("DEEPSEEK_API_KEY")
-global_s2_key = _read_runtime_secret("S2_API_KEY")
+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)
@@ -338,301 +253,268 @@ def inject_wechat_browser_notice() -> None:
     components.html(script, height=0, width=0)
 
 
-if "is_running" not in st.session_state: st.session_state.is_running = False
-if "start_pipeline" not in st.session_state: st.session_state.start_pipeline = False
-if "resume_pipeline" not in st.session_state: st.session_state.resume_pipeline = False
-if "zip_buffer" not in st.session_state: st.session_state.zip_buffer = None
-if "run_mode" not in st.session_state: st.session_state.run_mode = "amj"
-if "sp_generation_task" not in st.session_state: st.session_state.sp_generation_task = None
-if "sp_generation_idempotency_key" not in st.session_state: st.session_state.sp_generation_idempotency_key = ""
-if "sp_last_task_heartbeat" not in st.session_state: st.session_state.sp_last_task_heartbeat = 0.0
-if "sp_current_node" not in st.session_state: st.session_state.sp_current_node = ""
-if "sp_pending_task_finalization" not in st.session_state: st.session_state.sp_pending_task_finalization = None
-if "sp_pending_package" not in st.session_state: st.session_state.sp_pending_package = None
-if "sp_terminal_notice" not in st.session_state: st.session_state.sp_terminal_notice = None
-
 
-def _safe_user_workspace_key(value):
-    cleaned = re.sub(r"[^A-Za-z0-9_.-]", "_", str(value or "").strip())
-    return cleaned[:80] or "anonymous"
+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,
+}.items():
+    if key not in st.session_state:
+        st.session_state[key] = default
 
 
-def _load_recoverable_checkpoint():
-    """Return a validated checkpoint for the current authenticated user only."""
-    if not os.path.isfile(CHECKPOINT_FILE):
-        return None
-    try:
-        with open(CHECKPOINT_FILE, "r", encoding="utf-8") as handle:
-            state = json.load(handle)
-    except Exception:
-        return None
-    required = {
-        "topic", "keywords", "target_lang", "words",
-        "agent_state", "adaptive_structure", "completed_sections",
-    }
-    if not isinstance(state, dict) or not required.issubset(state):
-        return None
-    if not isinstance(state.get("adaptive_structure"), list):
-        return None
-    if not isinstance(state.get("completed_sections"), list):
-        return None
-    return state
+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 _prepare_new_generation_request() -> None:
-    st.session_state["sp_generation_idempotency_key"] = _new_generation_idempotency_key()
-    st.session_state["sp_generation_task"] = None
-    st.session_state["sp_last_task_heartbeat"] = 0.0
-    st.session_state["sp_current_node"] = "task_start"
-    st.session_state["zip_buffer"] = None
-    st.session_state["zip_filename"] = ""
-    st.session_state["sp_pending_task_finalization"] = None
-    st.session_state["sp_pending_package"] = None
-    st.session_state["sp_terminal_notice"] = None
-
-
-def _begin_managed_generation_task(*, is_resume: bool, loaded_state: dict | None, mode: str, language: str):
-    checkpoint_key = ""
-    if is_resume and isinstance(loaded_state, dict):
-        checkpoint_key = str(loaded_state.get("managed_task_idempotency_key") or "").strip()
-
-    idempotency_key = checkpoint_key or str(
-        st.session_state.get("sp_generation_idempotency_key") or ""
-    ).strip()
-    if not idempotency_key:
-        idempotency_key = _new_generation_idempotency_key()
-
-    response = start_generation_task(
-        idempotency_key=idempotency_key,
-        mode=mode,
-        interface_language=language,
-    )
-    task = dict(response.get("task") or {})
-    if not task.get("task_id"):
-        raise RuntimeError("task_start_missing_task_id")
+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)
 
-    st.session_state["sp_generation_idempotency_key"] = idempotency_key
-    st.session_state["sp_generation_task"] = task
-    st.session_state["sp_last_task_heartbeat"] = time.time()
-    logging.info(
-        "[SP_TASK] task_started task_id=%s mode=%s language=%s",
-        str(task.get("task_id") or ""),
-        mode,
-        language,
-    )
-    return task, idempotency_key
 
+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_output_dir(task_id: str) -> str:
-    root = globals().get("_user_workspace") or os.path.join(WORKSPACE_DIR, "users", "anonymous")
-    path = os.path.join(root, "tasks", _safe_user_workspace_key(task_id), "outputs")
-    os.makedirs(path, exist_ok=True)
-    return path
 
+def _task_dir(task_id: str) -> Path:
+    return Path(_user_workspace) / "tasks" / _safe_key(task_id)
 
-def _checkpoint_state_for_update() -> dict:
-    if not os.path.isfile(CHECKPOINT_FILE):
-        return {}
-    try:
-        with open(CHECKPOINT_FILE, "r", encoding="utf-8") as handle:
-            value = json.load(handle)
-        return value if isinstance(value, dict) else {}
-    except Exception:
-        return {}
 
+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 _write_checkpoint_state(value: dict) -> None:
-    if not isinstance(value, dict) or not value:
-        return
-    os.makedirs(os.path.dirname(CHECKPOINT_FILE), exist_ok=True)
-    temporary = CHECKPOINT_FILE + ".tmp"
-    with open(temporary, "w", encoding="utf-8") as handle:
-        json.dump(value, handle, ensure_ascii=False)
-    os.replace(temporary, CHECKPOINT_FILE)
 
+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")
 
-def _persist_pending_finalization() -> None:
-    pending = st.session_state.get("sp_pending_task_finalization")
-    if not isinstance(pending, dict):
-        return
-    state = _checkpoint_state_for_update()
-    if not state:
-        return
-    state["pending_task_finalization"] = pending
-    package = st.session_state.get("sp_pending_package")
-    if isinstance(package, dict):
-        state["pending_package"] = package
-    _write_checkpoint_state(state)
+    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"]),
+        "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 _restore_pending_finalization() -> None:
-    if isinstance(st.session_state.get("sp_pending_task_finalization"), dict):
-        return
-    state = _checkpoint_state_for_update()
-    pending = state.get("pending_task_finalization") if isinstance(state, dict) else None
-    if not isinstance(pending, dict):
-        return
-    st.session_state["sp_pending_task_finalization"] = pending
-    package = state.get("pending_package")
-    if isinstance(package, dict):
-        st.session_state["sp_pending_package"] = package
-    task_id = str(pending.get("task_id") or "").strip()
-    if task_id and not st.session_state.get("sp_generation_task"):
-        st.session_state["sp_generation_task"] = {"task_id": task_id}
+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 _remove_pending_checkpoint_metadata(*, delete_checkpoint=False, detach_task=False) -> None:
-    if delete_checkpoint:
-        try:
-            os.remove(CHECKPOINT_FILE)
-        except OSError:
-            pass
-        return
-    state = _checkpoint_state_for_update()
-    if not state:
-        return
-    state.pop("pending_task_finalization", None)
-    state.pop("pending_package", None)
-    if detach_task:
-        state["managed_task_idempotency_key"] = ""
-        state["managed_task_id"] = ""
-    _write_checkpoint_state(state)
+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 _prune_user_output_packages(max_files: int = 5) -> None:
-    root = globals().get("_user_workspace")
-    if not root or not os.path.isdir(root):
-        return
-    packages = []
-    for current_root, _, names in os.walk(os.path.join(root, "tasks")):
-        for name in names:
-            if name.endswith(".zip") and not name.startswith(".pending_"):
-                path = os.path.join(current_root, name)
+
+def _launch_executor(task_id: str, *, force: bool = False) -> int:
+    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)
+    marker_path = task_dir / "executor_launch.json"
+    if marker_path.is_file():
+        try:
+            marker = json.loads(marker_path.read_text(encoding="utf-8"))
+            pid = int(marker.get("pid") or 0)
+            verified_executor = _pid_is_task_executor(pid, task_id, task_dir)
+            if verified_executor and not force:
+                return pid
+            if verified_executor and force:
+                # Resume means D1 has already made the previous execution
+                # reclaimable. Stop only a PID whose /proc command line proves
+                # it is this task's old executor; a new epoch will be the sole
+                # authority after claim.
                 try:
-                    packages.append((os.path.getmtime(path), path))
+                    os.kill(pid, signal.SIGTERM)
                 except OSError:
                     pass
-    packages.sort()
-    while len(packages) > max_files:
-        _, old_path = packages.pop(0)
-        try:
-            os.remove(old_path)
-        except OSError:
+        except Exception:
             pass
+    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),
+    ]
+    with log_path.open("ab") as log_handle:
+        process = subprocess.Popen(
+            command,
+            cwd=str(Path(__file__).resolve().parent),
+            stdin=subprocess.DEVNULL,
+            stdout=log_handle,
+            stderr=subprocess.STDOUT,
+            close_fds=True,
+            start_new_session=True,
+            env=os.environ.copy(),
+        )
+    _atomic_json(marker_path, {"pid": process.pid, "launched_at": datetime.now(timezone.utc).isoformat()})
+    logging.info("[SP_UI] executor_launched task_id=%s pid=%s", task_id, process.pid)
+    return int(process.pid)
 
 
-def _process_pending_task_finalization():
-    pending = st.session_state.get("sp_pending_task_finalization")
-    if not isinstance(pending, dict):
-        return None
-    task_id = str(pending.get("task_id") or "").strip()
-    status = str(pending.get("status") or "").strip().lower()
-    if not task_id or status not in {"completed", "failed", "cancelled", "resumable"}:
-        raise RuntimeError("invalid_pending_task_finalization")
-
-    package_bytes = None
-    package = st.session_state.get("sp_pending_package")
-    if status == "completed":
-        if not isinstance(package, dict):
-            raise RuntimeError("validated_package_metadata_missing")
-        pending_path = str(package.get("pending_path") or "").strip()
-        if not pending_path or not os.path.isfile(pending_path):
-            raise RuntimeError("validated_package_file_missing")
-        with open(pending_path, "rb") as handle:
-            package_bytes = handle.read()
-        if not package_bytes:
-            raise RuntimeError("validated_package_file_empty")
-
-    logging.info(
-        "[SP_TASK] task_finalization_request task_id=%s status=%s node=%s",
-        task_id,
-        status,
-        str(pending.get("current_node") or "")[:120],
-    )
-    response = finish_generation_task(
-        task_id,
-        status=status,
-        safe_error_code=str(pending.get("safe_error_code") or "")[:80],
-        safe_error_detail=str(pending.get("safe_error_detail") or "")[:800],
-        current_node=str(pending.get("current_node") or "")[:240],
-    )
+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"):
+        shutil.rmtree(_task_dir(task_id), ignore_errors=True)
+    st.session_state["sp_pending_cancel_task_id"] = ""
+    st.session_state["sp_generation_task"] = dict(result.get("task") or {}) or None
+    st.rerun()
 
-    logging.info(
-        "[SP_TASK] task_finalization_confirmed task_id=%s status=%s reservation=%s",
-        task_id,
-        status,
-        str((response.get("reservation") or {}).get("status") or ""),
-    )
 
-    if status == "completed":
-        filename = str(package.get("filename") or "SmartPaper.zip")
-        pending_path = str(package.get("pending_path") or "")
-        final_path = os.path.join(_task_output_dir(task_id), os.path.basename(filename))
-        try:
-            os.replace(pending_path, final_path)
-        except OSError:
-            # The bytes were read and validated before quota commit, so the user
-            # can still download even if the local archival rename fails.
-            final_path = pending_path
-        st.session_state["zip_buffer"] = package_bytes
-        st.session_state["zip_filename"] = os.path.basename(filename)
-        st.session_state["sp_terminal_notice"] = {"kind": "success"}
-        _prune_user_output_packages()
-    elif status in {"failed", "cancelled"}:
+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" if status == "failed" else "warning",
-            "message": str(pending.get("safe_error_detail") or "")[:800],
+            "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.rerun()
 
-    _remove_pending_checkpoint_metadata(
-        delete_checkpoint=bool(pending.get("delete_checkpoint")),
-        detach_task=bool(pending.get("detach_checkpoint")),
-    )
-    if status == "resumable":
-        st.session_state["sp_generation_task"] = dict(response.get("task") or {"task_id": task_id})
-    else:
-        st.session_state["sp_generation_task"] = None
-        st.session_state["sp_generation_idempotency_key"] = ""
-        st.session_state["sp_last_task_heartbeat"] = 0.0
-    st.session_state["sp_pending_task_finalization"] = None
-    st.session_state["sp_pending_package"] = None
-    return response
-
-
-def _finish_managed_generation_task(
-    status: str,
-    *,
-    safe_error_code: str = "",
-    safe_error_detail: str = "",
-    current_node: str = "",
-    delete_checkpoint: bool = False,
-    detach_checkpoint: bool = False,
-):
-    task = st.session_state.get("sp_generation_task") or {}
+
+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:
-        return None
-    pending = {
-        "task_id": task_id,
-        "status": str(status or "").strip().lower(),
-        "safe_error_code": str(safe_error_code or "")[:80],
-        "safe_error_detail": str(safe_error_detail or "")[:800],
-        "current_node": str(current_node or "")[:240],
-        "delete_checkpoint": bool(delete_checkpoint),
-        "detach_checkpoint": bool(detach_checkpoint),
-    }
-    st.session_state["sp_pending_task_finalization"] = pending
-    _persist_pending_finalization()
-    return _process_pending_task_finalization()
-
-
-def _quota_start_error_message(language: str, error: Exception) -> str:
-    raw = str(error or "")
-    return tr(language, "quota_exhausted") if "payment_required" in raw else raw
+        st.session_state["sp_pending_launch"] = None
+        st.error("task_start_missing_task_id")
+        return
+    st.session_state["sp_generation_task"] = task
+    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.rerun()
 
 
 inject_wechat_browser_notice()
@@ -642,8 +524,10 @@ DEFAULT_KEYWORDS = "Ubiquitous computing; Context awareness; Deep learning; Pers
 
 _fragment = getattr(st, "fragment", None) or getattr(st, "experimental_fragment", None)
 if _fragment is None:
-    def _fragment(function):
-        return function
+    def _fragment(function=None, **_kwargs):
+        if function is not None:
+            return function
+        return lambda fn: fn
 
 
 def _current_language_code() -> str:
@@ -651,777 +535,161 @@ def _current_language_code() -> str:
     return LANGUAGE_OPTIONS.get(choice, "zh")
 
 
-# Public access uses Alipay only. Password, legacy, and pilot login paths are not exposed.
 authenticated_user = require_alipay_login(_current_language_code())
-
-# Checkpoints are user-scoped. The former global snapshot.json could cause one
-# user or an old run to trigger a false recovery notice for another session.
-_user_workspace = os.path.join(
-    WORKSPACE_DIR,
-    "users",
-    _safe_user_workspace_key(authenticated_user.get("id")),
-)
+_user_workspace = os.path.join(WORKSPACE_DIR, "users", _safe_key(authenticated_user.get("id")))
 os.makedirs(_user_workspace, exist_ok=True)
-CHECKPOINT_FILE = os.path.join(_user_workspace, "snapshot.json")
-_restore_pending_finalization()
-_process_pending_task_finalization()
 
 
 @_fragment
 def render_workspace_controls(user):
-    """Render controls in an isolated rerun scope for fast language switching."""
     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",
+                "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)
 
-    # Cross-device global busy check remains unchanged in function, but all visible text is localized.
-    is_my_session = st.session_state.get("is_running", False) or \
-                    st.session_state.get("start_pipeline", False) or \
-                    st.session_state.get("resume_pipeline", False)
-    ui_blocked = (not is_my_session and is_system_busy())
-    st.session_state["sp_ui_blocked"] = ui_blocked
-    if ui_blocked:
-        st.markdown(
-            f"<h3 style='text-align:center;color:#e6a23c;'>{tr(lang_code, 'system_busy_title')}</h3>",
-            unsafe_allow_html=True,
-        )
-        st.warning(tr(lang_code, "system_busy_warning"))
-        st.info(tr(lang_code, "system_busy_info"))
-        return
+    task = st.session_state.get("sp_generation_task") or {}
+    task_status = str(task.get("status") or "")
+    active = task_status in {"created", "running", "resumable", "cancelling"}
+    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 _load_recoverable_checkpoint() is not None and not (st.session_state.is_running or st.session_state.resume_pipeline or st.session_state.start_pipeline):
+    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"):
-            if is_system_busy():
-                st.warning(tr(lang_code, "busy_short"))
-            else:
-                checkpoint_state = _load_recoverable_checkpoint() or {}
-                checkpoint_key = str(checkpoint_state.get("managed_task_idempotency_key") or "").strip()
-                st.session_state["sp_generation_idempotency_key"] = checkpoint_key or _new_generation_idempotency_key()
-                st.session_state["sp_generation_task"] = None
-                st.session_state["sp_last_task_heartbeat"] = 0.0
-                update_heartbeat()
-                st.session_state.resume_pipeline = True
-                st.session_state.start_pipeline = False
-                st.rerun()
-
+            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"):
-            if is_system_busy():
-                st.warning(tr(lang_code, "busy_short"))
-            else:
-                checkpoint_state = _load_recoverable_checkpoint() or {}
-                checkpoint_task_id = str(checkpoint_state.get("managed_task_id") or "").strip()
-                if checkpoint_task_id:
-                    st.session_state["sp_generation_task"] = {"task_id": checkpoint_task_id}
-                    try:
-                        _finish_managed_generation_task(
-                            "cancelled",
-                            safe_error_code="E_USER_CLEARED_CHECKPOINT",
-                            safe_error_detail="User cleared the recoverable checkpoint.",
-                            current_node="checkpoint_clear",
-                            delete_checkpoint=True,
-                        )
-                    except Exception as exc:
-                        st.error(str(exc))
-                        return
-                else:
-                    try:
-                        os.remove(CHECKPOINT_FILE)
-                    except OSError:
-                        pass
-                clear_heartbeat()
-                st.rerun()
+            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=st.session_state.is_running,
-        key="sp_topic",
-    )
-    st.text_input(
-        tr(lang_code, "keywords_label"),
-        value=DEFAULT_KEYWORDS,
-        disabled=st.session_state.is_running,
-        key="sp_keywords",
-    )
+    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=12000,
-        max_value=25000,
-        value=20000,
+        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=st.session_state.is_running,
+        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=st.session_state.is_running,
-        )
-        st.file_uploader(
-            tr(lang_code, "upload_stats"),
-            type=["csv", "xlsx", "xls"],
-            key="up_amj_data",
-            disabled=st.session_state.is_running,
-        )
-
+        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"))
-            amj_csv_template = get_amj_csv_template(lang_code)
-            st.code(amj_csv_template, language="csv")
-            st.download_button(
-                label=tr(lang_code, "download_amj_template"),
-                data=amj_csv_template,
-                file_name="AMJ_Data_Template.csv",
-                mime="text/csv",
-                use_container_width=True,
-            )
-
-        if st.button(tr(lang_code, "start_amj"), disabled=st.session_state.is_running, key="btn_amj"):
-            if is_system_busy():
-                st.warning(tr(lang_code, "busy_short"))
-            else:
-                _prepare_new_generation_request()
-                update_heartbeat()
-                st.session_state.run_mode = "amj"
-                st.session_state.start_pipeline = True
-                st.rerun()
+            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)
+        if 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.rerun()
 
     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=st.session_state.is_running,
-        )
-        st.file_uploader(
-            tr(lang_code, "upload_interviews"),
-            type=["txt", "pdf", "doc", "docx"],
-            key="up_asq_data",
-            disabled=st.session_state.is_running,
-        )
-
+        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"))
-            asq_txt_template = get_asq_txt_template(lang_code)
-            st.code(asq_txt_template, language="text")
-            st.download_button(
-                label=tr(lang_code, "download_asq_template"),
-                data=asq_txt_template,
-                file_name="ASQ_Interview_Template.txt",
-                mime="text/plain",
-                use_container_width=True,
-            )
+            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)
+        if 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.rerun()
 
-        if st.button(tr(lang_code, "start_asq"), disabled=st.session_state.is_running, key="btn_asq"):
-            if is_system_busy():
-                st.warning(tr(lang_code, "busy_short"))
-            else:
-                _prepare_new_generation_request()
-                update_heartbeat()
-                st.session_state.run_mode = "asq"
-                st.session_state.start_pipeline = True
-                st.rerun()
-
-    # Localize Streamlit built-in uploader text after the uploaders are rendered.
     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())
-target_lang = str(st.session_state.get("sp_target_lang") or get_target_language(lang_code))
-output_suffix = str(st.session_state.get("sp_output_suffix") or get_output_suffix(lang_code))
-target_flag = str(st.session_state.get("sp_target_flag") or get_target_flag(lang_code))
-
-topic = str(st.session_state.get("sp_topic") or DEFAULT_TOPIC)
-keywords = str(st.session_state.get("sp_keywords") or DEFAULT_KEYWORDS)
-words = int(st.session_state.get("sp_words") or 20000)
-uploaded_papers_amj = st.session_state.get("up_amj_ref") or []
-uploaded_data_amj = st.session_state.get("up_amj_data")
-uploaded_papers_asq = st.session_state.get("up_asq_ref") or []
-uploaded_interviews = st.session_state.get("up_asq_data")
-
-if st.session_state.get("sp_ui_blocked"):
-    st.stop()
-
-if st.session_state.get("sp_payment_required") and not (st.session_state.start_pipeline or st.session_state.resume_pipeline):
+
+_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)
 
-_terminal_notice = st.session_state.pop("sp_terminal_notice", None)
-if isinstance(_terminal_notice, dict) and _terminal_notice.get("kind") in {"error", "warning"}:
-    _message = str(_terminal_notice.get("message") or "").strip()
-    if _message:
-        (st.error if _terminal_notice.get("kind") == "error" else st.warning)(_message)
-
-progress_bar = st.empty()
-monitor_ui = st.empty()
-
-def render_monitor(agent, phase, current_sec="", details=""):
-    node = str(current_sec or phase or "")[:240]
-    st.session_state["sp_current_node"] = node
-    if node and node != st.session_state.get("sp_last_logged_node"):
-        task_id = str((st.session_state.get("sp_generation_task") or {}).get("task_id") or "")
-        logging.info("[SP_TASK] node_enter task_id=%s node=%s", task_id, node)
-        st.session_state["sp_last_logged_node"] = node
-    monitor_ui.markdown(f"""
+
+def _render_monitor_box(task: dict) -> None:
+    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)))
+    if status in {"created", "running", "resumable", "cancelling"}:
+        st.progress(progress, text=f"{progress}% — {node}")
+    st.markdown(f"""
     <div class='log-box'>
-        <div><b>{tr(lang_code, 'monitor_pipeline')}:</b> {phase}</div><div><b>{tr(lang_code, 'monitor_node')}:</b> {current_sec}</div><div><b>{tr(lang_code, 'monitor_log')}:</b> {details}</div><hr>
+        <div><b>{tr(lang_code, 'monitor_pipeline')}:</b> {status}</div><div><b>{tr(lang_code, 'monitor_node')}:</b> {node}</div><div><b>{tr(lang_code, 'monitor_log')}:</b> checkpoint {int(task.get('checkpoint_seq') or 0)} · {str(task.get('checkpoint_stage') or '')}</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>{agent.ds_calls}</b></span>   
-            <span style='background: #dbeafe; padding: 4px 8px; border-radius: 4px; color: black;'>📚 S2: <b>{agent.s2_calls}</b></span>
-            <span style='background: #d1fae5; padding: 4px 8px; border-radius: 4px; color: black;'>🌐 OpenAlex: <b>{agent.oa_calls}</b></span>
+            <span style='background: #fef3c7; padding: 4px 8px; border-radius: 4px; color: black;'>🧠 DeepSeek: <b>{int(task.get('ds_calls') or 0)}</b></span>
+            <span style='background: #dbeafe; padding: 4px 8px; border-radius: 4px; color: black;'>📚 S2: <b>{int(task.get('s2_calls') or 0)}</b></span>
+            <span style='background: #d1fae5; padding: 4px 8px; border-radius: 4px; color: black;'>🌐 OpenAlex: <b>{int(task.get('oa_calls') or 0)}</b></span>
         </div>
     </div>
     """, unsafe_allow_html=True)
 
-FLAG_MAP = {"Simplified Chinese (简体中文)": "🇨🇳", "Russian (Русский)": "🇷🇺", "Kazakh (Қазақша)": "🇰🇿"}
-
-if st.session_state.start_pipeline or st.session_state.resume_pipeline:
-    if not global_ds_key: 
-        st.error(tr(lang_code, "api_key_missing"))
-        st.stop()
-        
-    if st.session_state.get("is_running", False):
-        recoverable_state = _load_recoverable_checkpoint()
-        st.session_state.is_running = False
-        st.session_state.start_pipeline = False
-        st.session_state.resume_pipeline = bool(recoverable_state)
-        if recoverable_state:
-            st.toast(tr(lang_code, "resume_toast"), icon="⚡")
-            update_heartbeat()
-            st.rerun()
-        # A normal page rerun without a valid checkpoint is not a recovery event.
-        # Keep the already-rendered writing interface and show no recovery toast.
-        clear_heartbeat()
-        st.stop()
-
-    is_resume = st.session_state.resume_pipeline
-    loaded_state = _load_recoverable_checkpoint() if is_resume else None
-    if is_resume and loaded_state is None:
-        is_resume = False
-
-    try:
-        managed_task, managed_task_key = _begin_managed_generation_task(
-            is_resume=is_resume,
-            loaded_state=loaded_state,
-            mode=st.session_state.run_mode,
-            language=lang_code,
-        )
-    except Exception as exc:
-        clear_heartbeat()
-        st.session_state.is_running = False
-        st.session_state.start_pipeline = False
-        st.session_state.resume_pipeline = False
-        st.error(_quota_start_error_message(lang_code, exc))
-        if "payment_required" in str(exc):
-            st.session_state["sp_payment_required"] = True
-            render_payment_purchase(lang_code)
-        st.stop()
-
-    st.session_state.is_running = True
-    agent = SSciPipelineAgent(global_ds_key, global_s2_key)
-    agent.set_execution_progress_hook(update_heartbeat)
-    SAFE_CHUNK_WORDS = agent.chunk_words_default
-    qual_engine = get_qualitative_engine(agent)
-    quant_engine = get_quantitative_engine()
-    completion_finalization_attempted = False
 
-    try:
-        update_heartbeat()
-                
-        if is_resume and loaded_state:
-            pb = progress_bar.progress(50, text=tr(lang_code, "resume_loading"))
-            state = loaded_state
-            state["managed_task_idempotency_key"] = managed_task_key
-            state["managed_task_id"] = managed_task.get("task_id", "")
-            topic = state["topic"]
-            keywords = state["keywords"]
-            target_lang = state["target_lang"]
-            words = state["words"]
-            user_data_md = state["user_data_md"]
-            data_metadata = state.get("data_metadata", {})
-            user_lit_context = state["user_lit_context"]
-            
-            agent.load_state(state["agent_state"])
-            adaptive_structure = state["adaptive_structure"]
-            completed_sections = state["completed_sections"]
-            full_draft_en = state["full_draft_en"]
-            full_draft_translated = state["full_draft_translated"]
-            synthesis_result = state["synthesis_result"]
-            refs_text = state["refs_text"]
-            context_tail = state["context_tail"]
-            
-            st.session_state.run_mode = state.get("run_mode", "amj")
-            if st.session_state.run_mode == "amj":
-                APP_CONFIG.update(AMJ_PROMPTS)
-            else:
-                APP_CONFIG.update(ASQ_PROMPTS)
-                
-            st.success(tr(lang_code, "resume_success", done=len(completed_sections), total=len(adaptive_structure)))
-            render_monitor(agent, tr(lang_code, "resume_phase"), tr(lang_code, "state_check"), tr(lang_code, "history_restored"))
-            
+def _render_task_status_once() -> 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 {"created", "running", "resumable", "cancelling"}:
+        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__)
+    _render_monitor_box(task)
+    if 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:
-            pb = progress_bar.progress(1, text=tr(lang_code, "pipeline_start"))
-            user_lit_context = ""
-            user_data_md = ""
-            data_metadata = {}
-            
-            if st.session_state.run_mode == "amj":
-                if uploaded_papers_amj:
-                    user_lit_context = extract_text_from_files(uploaded_papers_amj[:20])
-                    with st.status(tr(lang_code, "literature_extract"), expanded=True) as status:
-                        ph = st.empty()
-                        all_paper_texts = [extract_text_from_files([f]) for f in uploaded_papers_amj[:20]]
-                        
-                        future = ThreadPoolExecutor(max_workers=1).submit(extract_golden_metadata, agent, all_paper_texts)
-                        dot_count = 0
-                        while not future.done():
-                            update_heartbeat()
-                            ph.markdown(tr(lang_code, "literature_deep") + " ⏳" * (dot_count % 4))
-                            time.sleep(1)
-                            dot_count += 1
-                            
-                        agent.global_memory["core_golden_metadata"] = future.result()
-                        st.session_state.rag_engine = initialize_rag(all_paper_texts)
-                        status.update(label=tr(lang_code, "literature_done"), state="complete", expanded=False)
-                
-                with st.spinner(tr(lang_code, "extract_quant")):
-                    update_heartbeat()
-                    _, user_data_md, msg, data_metadata = quant_engine.resolve_empirical_data(uploaded_data_amj, lang_code)
-                    st.info(msg)
-                    
-            elif st.session_state.run_mode == "asq":
-                if uploaded_papers_asq:
-                    user_lit_context = extract_text_from_files(uploaded_papers_asq[:20])
-                    with st.status(tr(lang_code, "literature_extract"), expanded=True) as status:
-                        ph = st.empty()
-                        all_paper_texts = [extract_text_from_files([f]) for f in uploaded_papers_asq[:20]]
-                        
-                        future = ThreadPoolExecutor(max_workers=1).submit(extract_golden_metadata, agent, all_paper_texts)
-                        dot_count = 0
-                        while not future.done():
-                            update_heartbeat()
-                            ph.markdown(tr(lang_code, "literature_deep") + " ⏳" * (dot_count % 4))
-                            time.sleep(1)
-                            dot_count += 1
-                            
-                        agent.global_memory["core_golden_metadata"] = future.result()
-                        st.session_state.rag_engine = initialize_rag(all_paper_texts)
-                        status.update(label=tr(lang_code, "literature_done"), state="complete", expanded=False)
-                
-                update_heartbeat()
-                raw_interviews = extract_text_from_files([uploaded_interviews]) if uploaded_interviews else ""
-                user_data_md, msg, data_metadata = qual_engine.resolve_qualitative_data(topic, keywords, raw_interviews)
-                st.info(msg or tr(lang_code, "qualitative_ready"))
-
-            state = {
-                "topic": topic, "keywords": keywords, "target_lang": target_lang, "words": words,
-                "user_data_md": user_data_md, "data_metadata": data_metadata, "user_lit_context": user_lit_context,
-                "completed_sections": [], "full_draft_en": "", "full_draft_translated": "",
-                "context_tail": "", "run_mode": st.session_state.run_mode, "interface_language": lang_code,
-                "managed_task_idempotency_key": managed_task_key,
-                "managed_task_id": managed_task.get("task_id", ""),
-            }
-
-            pb.progress(5, text=tr(lang_code, "config_route"))
-            
-            if st.session_state.run_mode == "amj":
-                APP_CONFIG.update(AMJ_PROMPTS)
-                state["prompt_config"] = AMJ_PROMPTS
-            else:
-                APP_CONFIG.update(ASQ_PROMPTS)
-                state["prompt_config"] = ASQ_PROMPTS
-
-            pb.progress(10, text=tr(lang_code, "literature_search"))
-            def sync_monitor_ui():
-                update_heartbeat()
-                render_monitor(agent, tr(lang_code, "search_phase"), "S2 + OpenAlex", tr(lang_code, "search_detail"))
-            
-            target_refs_count = 50
-            refs_text = agent.fetch_references_parallel(topic, keywords, min_required=target_refs_count, progress_callback=sync_monitor_ui)
-            
-            if len(agent.citation_map) < target_refs_count:
-                raise RuntimeError(
-                    tr(lang_code, "refs_insufficient", count=len(agent.citation_map), required=target_refs_count)
-                )
-            else: 
-                st.success(tr(lang_code, "refs_success", count=len(agent.citation_map)))
-
-            with st.spinner(tr(lang_code, "rag_sync")):
-                update_heartbeat()
-                if hasattr(st.session_state, "rag_engine") and st.session_state.rag_engine is not None:
-                    st.session_state.rag_engine._initialize_chunks([refs_text])
-                else:
-                    st.session_state.rag_engine = initialize_rag([refs_text])
-            st.success(tr(lang_code, "rag_done"))
-
-            if user_lit_context:
-                pb.progress(11, text=tr(lang_code, "core_literature_parsing"))
-                update_heartbeat()
-                core_refs = agent.parse_user_core_literature(user_lit_context)
-                if core_refs:
-                    refs_text = "【USER PROVIDED CORE LITERATURE】\n" + core_refs + "\n\n【RETRIEVED LITERATURE】\n" + refs_text
-                    st.success(tr(lang_code, "core_refs_done"))
-
-            state["refs_text"] = refs_text
-
-            pb.progress(12, text=tr(lang_code, "synthesis_progress"))
-            with st.status(tr(lang_code, "synthesis_status"), expanded=True) as status:
-                ph = st.empty()
-                synthesis_result = ph.write_stream(stream_with_heartbeat(agent.stream_synthesize_literature(topic, refs_text, user_core_literature=user_lit_context)))
-                synthesis_result = re.sub(r'<think>.*?</think>', '', synthesis_result, flags=re.DOTALL).strip()
-                status.update(label=tr(lang_code, "synthesis_done"), state="complete", expanded=False)
-            state["synthesis_result"] = synthesis_result
-            
-            pb.progress(14, text=tr(lang_code, "anchor_progress"))
-            with st.status(tr(lang_code, "anchor_status"), expanded=True) as status:
-                ph = st.empty()
-                anchor_response = ph.write_stream(stream_with_heartbeat(agent.stream_generate_anchors(topic, keywords, synthesis_result, refs_text, data_evidence=user_data_md, mode=st.session_state.run_mode)))
-                status.update(label=tr(lang_code, "anchor_done"), state="complete", expanded=False)
-            
-            # ==========================================================
-            # [核心修复] 必须先剔除 <think> 标签！防止正则匹配到模型在思考时引用提示词的空占位符
-            # ==========================================================
-            anchor_response_clean = re.sub(r'<think>.*?</think>', '', anchor_response, flags=re.DOTALL).strip()
-            
-            abstract_match = re.search(r'-+\s*ABSTRACT\s*-+\s*(.*?)(?=\s*-+\s*CONCLUSION\s*-+|$)', anchor_response_clean, re.DOTALL | re.IGNORECASE)
-            conclusion_match = re.search(r'-+\s*CONCLUSION\s*-+\s*(.*?)$', anchor_response_clean, re.DOTALL | re.IGNORECASE)
-            abstract_anchor = abstract_match.group(1).strip() if abstract_match else anchor_response_clean.strip()
-            conclusion_anchor = conclusion_match.group(1).strip() if conclusion_match else ""
-            agent.global_memory["abstract_anchor"] = abstract_anchor
-            agent.global_memory["conclusion_anchor"] = conclusion_anchor
-            # ==========================================================
-            
-            pb.progress(15, text=tr(lang_code, "outline_progress"))
-            update_heartbeat()
-            plan = agent.generate_outline(topic, keywords, mode=st.session_state.run_mode, data_metadata=data_metadata)
-            en_topic = str(plan.get("en_topic", topic))
-            structure = plan.get("structure", [])
-
-            en_abstract_clean = clean_markdown_headings(re.sub(r'(?i)^abstract[:\s]*', '', str(abstract_anchor)).strip())
-            trans_abstract_clean = ""
-
-            loc_abs = {"Simplified Chinese (简体中文)": "摘要", "Russian (Русский)": "Аннотация", "Kazakh (Қазақша)": "Түйіндеме"}.get(target_lang, "Abstract")
-            loc_kw = {"Simplified Chinese (简体中文)": "关键词", "Russian (Русский)": "Ключевые слова", "Kazakh (Қазақша)": "Кілт сөздер"}.get(target_lang, "Keywords")
-
-            full_draft_en = f"# {en_topic}\n\n## Abstract\n{en_abstract_clean}\n\n## Keywords\n{keywords}\n\n"
-            full_draft_translated = f"# {topic}\n\n## {loc_abs}\n{trans_abstract_clean}\n\n## {loc_kw}\n{keywords}\n\n"
-
-            adaptive_structure = agent.build_generation_tasks(structure, int(words), SAFE_CHUNK_WORDS)
-            for task in adaptive_structure:
-                task["is_first"] = bool(task.get("section_first"))
-
-            with st.spinner(tr(lang_code, "jit_status")):
-                update_heartbeat()
-                jit_texts = agent.prefetch_jit_literature(topic, keywords, adaptive_structure)
-                if jit_texts:
-                    if hasattr(st.session_state, "rag_engine") and st.session_state.rag_engine is not None:
-                        st.session_state.rag_engine._initialize_chunks([jit_texts])
-                    else:
-                        st.session_state.rag_engine = initialize_rag([jit_texts])
-                    st.success(tr(lang_code, "jit_done"))
-
-            state["adaptive_structure"] = adaptive_structure
-            state["agent_state"] = agent.export_state()
-            state["full_draft_en"] = full_draft_en
-            state["full_draft_translated"] = full_draft_translated
-            completed_sections = []
-            context_tail = ""
-            
-            os.makedirs(WORKSPACE_DIR, exist_ok=True)
-            with open(CHECKPOINT_FILE, "w", encoding="utf-8") as f: 
-                json.dump(state, f, ensure_ascii=False)
-
-        total_steps = len(adaptive_structure)
-        current_step = len(completed_sections)
-        st.markdown("---")
-        ui_blocks = []
-        target_flag = FLAG_MAP.get(target_lang, "🌐")
-
-        for sec in adaptive_structure:
-            display_name = sec["display_name"]
-            st.markdown(f"### ⚙️ {display_name}")
-            col_en, col_trans = st.columns(2)
-            with col_en:
-                st.markdown(tr(lang_code, "english_draft"))
-                en_ph = st.empty()
-            with col_trans:
-                st.markdown(tr(lang_code, "target_draft", flag=target_flag))
-                trans_ph = st.empty()
-            if display_name in completed_sections:
-                en_ph.success(tr(lang_code, "restored"))
-                trans_ph.info(tr(lang_code, "translation_waiting"))
-            else:
-                en_ph.info(tr(lang_code, "waiting"))
-                trans_ph.info(tr(lang_code, "translation_waiting"))
-            ui_blocks.append({"sec": sec, "en_ph": en_ph, "trans_ph": trans_ph})
-
-        generation_aborted = False
-        generation_failure_code = ""
-        generation_failure_detail = ""
-        accumulated_deviation = 0
-        remaining_sections = sum(1 for item in adaptive_structure if item["display_name"] not in completed_sections)
-        all_cite_keys = list(agent.citation_map.keys())
-        valid_chunks = [item for item in adaptive_structure if not any(word in item["section"].lower() for word in ["abstract","keyword","appendix"])]
-        allocation_map = {}
-        current_cite_idx = 0
-        for idx, item in enumerate(valid_chunks):
-            count = max(1, math.ceil(len(all_cite_keys) / max(1, len(valid_chunks))))
-            allocation_map[item["display_name"]] = all_cite_keys[current_cite_idx:] if idx == len(valid_chunks)-1 else all_cite_keys[current_cite_idx:current_cite_idx+count]
-            current_cite_idx += count
-
-        for block in ui_blocks:
-            if generation_aborted:
-                break
-            sec = block["sec"]
-            en_ph = block["en_ph"]
-            trans_ph = block["trans_ph"]
-            sec_title = sec["section"]
-            display_name = sec["display_name"]
-            if display_name in completed_sections:
-                continue
-            remaining_sections = max(0, remaining_sections - 1)
-            adjustment = -accumulated_deviation / remaining_sections if remaining_sections else 0
-            adjusted_target = min(int(SAFE_CHUNK_WORDS * 1.5), max(80, int(sec["target_words"] + adjustment)))
-            is_quarantined = any(word in sec_title.lower() for word in ["abstract","keyword","conclusion","references","appendix"])
-            assigned_tags = ", ".join(allocation_map.get(display_name, [])) or "the single most relevant source"
-            citation_rule = "NO IN-TEXT CITATIONS ALLOWED IN THIS SECTION." if is_quarantined else f"Use claim-specific citation. Each sentence may contain at most one source tag. Candidate tags: {assigned_tags}. <AVAILABLE_CITATIONS>{refs_text}</AVAILABLE_CITATIONS>"
-            memory_context = agent._build_memory_context(sec_title, 2500)
-            memory_block = f"\n<GLOBAL_MEMORY>\n{memory_context}\n</GLOBAL_MEMORY>\n" if memory_context else ""
-            if getattr(st.session_state, "rag_engine", None) is not None:
-                snippets = st.session_state.rag_engine.get_relevant_context(display_name)
-                if snippets:
-                    memory_block += f"\n<TARGETED_RAG>\n{snippets}\n</TARGETED_RAG>\n"
-            subsection = sec.get("subsection")
-            heading_directive = "Do not output the section heading; it will be inserted by the system."
-            if subsection:
-                heading_directive += f" Do not output the subsection heading '{subsection}'; it will also be inserted by the system. Use level-3 headings only when analytically necessary."
-            elif sec_title.lower() in {"introduction", "conclusion"}:
-                heading_directive += " Do not use additional headings."
-            else:
-                heading_directive += " Use level-2 or level-3 headings only when they clarify the argument; never use level-4 headings or numbered headings."
-            table_rule = "Present statistical evidence in standard Markdown tables with plain-text captions." if sec["is_method_or_result"] and st.session_state.run_mode == "amj" else ""
-            required_content = str(sec.get("required_content") or "")
-            concepts = ", ".join(sec.get("concepts") or [])
-            qualitative_rule = f"Assigned qualitative concepts: {concepts}. Use only their assigned representative quotations and do not repeat a quotation already used elsewhere." if concepts else ""
-            data_injection = f"\n<RESEARCH_EVIDENCE>\n{user_data_md}\n</RESEARCH_EVIDENCE>\n" if user_data_md and sec["is_method_or_result"] else ""
-            completed_label = ", ".join(completed_sections)
-            anti_repeat = f"Already completed: {completed_label}. Do not restate their background, gap, or contribution." if completed_label else ""
-            sys_prompt = APP_CONFIG.get("sys_academic_writer", "") + "\n" + APP_CONFIG.get("deep_refinement_directive", "")
-            user_prompt = f"""Write the assigned English manuscript unit.
-Section: {sec_title}
-Subsection: {subsection or 'None'}
-Target length: {int(adjusted_target*0.9)}-{int(adjusted_target*1.1)} words.
-{heading_directive}
-{citation_rule}
-{table_rule}
-{required_content}
-{qualitative_rule}
-{anti_repeat}
-{memory_block}
-Literature synthesis: {synthesis_result}
-{data_injection}
-Previous tail: {context_tail[-2500:]}
-Output only final academic prose. Do not report the word count."""
-            if agent._count_tokens(sys_prompt) + agent._count_tokens(user_prompt) > 60000:
-                user_prompt = agent._truncate_context(user_prompt, 60000 - agent._count_tokens(sys_prompt))
-            render_monitor(agent, tr(lang_code, "drafting_phase", current=current_step+1, total=total_steps), display_name, tr(lang_code, "target_words", words=adjusted_target, engine=tr(lang_code, "engine_amj") if st.session_state.run_mode=="amj" else tr(lang_code, "engine_asq")))
-            try:
-                with st.status(tr(lang_code, "draft_status", name=display_name), expanded=True) as status:
-                    update_heartbeat()
-                    en_raw = en_ph.write_stream(stream_with_heartbeat(agent.stream_deepseek(sys_prompt, user_prompt, max_tokens=7000, model_role="writing", thinking=False, allow_fallback=False)))
-                    status.update(label=tr(lang_code, "draft_done", name=display_name), state="complete", expanded=False)
-                en_raw = re.sub(r'<think>.*?</think>', '', str(en_raw), flags=re.S).strip()
-                en_raw = re.sub(rf'(?im)^#\s*{re.escape(sec_title)}\s*\n+', '', en_raw).strip()
-                if subsection:
-                    en_raw = re.sub(rf'(?im)^##\s*{re.escape(subsection)}\s*\n+', '', en_raw).strip()
-                en_raw = re.sub(r'(?m)^(#{2,3})\s+[\d\.\s一二三四五六七八九十、]+\s*(.*)', r'\1 \2', en_raw)
-                en_raw = re.sub(r'(?m)^#{4,}\s+(.*)', r'**\1**', en_raw)
-                en_raw = re.sub(r'(?m)^#\s+(.+)$', r'## \1', en_raw)
-                if st.session_state.run_mode == "asq" and user_data_md and "appendix" not in sec_title.lower():
-                    quote_dict = {tag: quote for quote, tag in re.findall(r'"([^"]+)"\s*\[(QUOTE-\d+)\]', user_data_md)}
-                    for tag, quote in quote_dict.items():
-                        en_raw = en_raw.replace(tag, f'"{quote}"')
-                en_raw = agent.polish_academic_section(en_raw, display_name, abstract=False)
-                prefix = ""
-                if sec.get("section_first"):
-                    prefix += f"# {sec_title}\n\n"
-                if sec.get("subsection_first") and subsection:
-                    prefix += f"## {subsection}\n\n"
-                en_with_headings = prefix + en_raw
-                en_resolved = agent.audit_and_resolve_citations(en_with_headings, section_name=display_name, quarantine=is_quarantined)
-                en_ph.markdown(en_resolved)
-                trans_ph.info(tr(lang_code, "translation_waiting"))
-                full_draft_en += en_resolved + "\n\n"
-                agent._extract_memory_from_text(en_raw, sec_title)
-                words_tail = en_raw.split()
-                context_tail = " ".join(words_tail[-200:])
-                actual_words = len(en_raw.split())
-                accumulated_deviation += actual_words - adjusted_target
-                current_step += 1
-                completed_sections.append(display_name)
-                state.update({
-                    "completed_sections": completed_sections,
-                    "full_draft_en": full_draft_en,
-                    "full_draft_translated": "",
-                    "context_tail": context_tail,
-                    "agent_state": agent.export_state(),
-                    "data_metadata": data_metadata,
-                })
-                with open(CHECKPOINT_FILE, "w", encoding="utf-8") as checkpoint:
-                    json.dump(state, checkpoint, ensure_ascii=False)
-                st.caption(tr(lang_code, "word_count", actual=actual_words, target=adjusted_target, deviation=accumulated_deviation))
-                pb.progress(min(15 + int((current_step/max(1,total_steps))*75), 90), text=tr(lang_code, "section_complete_progress", name=display_name))
-            except Exception as error:
-                en_ph.error(tr(lang_code, "node_error"))
-                st.error(tr(lang_code, "node_error_detail", error=error))
-                generation_aborted = True
-                generation_failure_code = "E_GENERATION_NODE_FAILED"
-                generation_failure_detail = str(error)[:800]
-
-        if generation_aborted:
-            _finish_managed_generation_task(
-                "failed",
-                safe_error_code=generation_failure_code or "E_GENERATION_INTERRUPTED",
-                safe_error_detail=generation_failure_detail or "Generation stopped; quota was released and the checkpoint was retained for a fresh retry.",
-                current_node=str(st.session_state.get("sp_current_node") or "generation"),
-                detach_checkpoint=True,
-            )
-
-        if not generation_aborted and full_draft_en.strip():
-            pb.progress(91, text=tr(lang_code, "pack_docx"))
-            update_heartbeat()
-            full_draft_en = clean_trailing_references(full_draft_en, ["Reference","References","Reference List"])
-            # The final abstract is derived from the completed English manuscript and frozen evidence.
-            final_abstract = agent.generate_final_abstract(topic, keywords, full_draft_en, data_evidence=user_data_md, mode=st.session_state.run_mode)
-            full_draft_en = replace_abstract(full_draft_en, final_abstract)
-            agent.validate_manuscript_structure(full_draft_en, adaptive_structure, translated=False)
-            if st.session_state.run_mode == "asq":
-                repeated_quotes = agent.find_repeated_qualitative_quotes(full_draft_en, data_metadata)
-                if repeated_quotes:
-                    raise RuntimeError(f"Repeated qualitative quotations: {repeated_quotes[:5]}")
-            # Only sources actually cited in the manuscript enter the reference list.
-            references = []
-            seen_refs = set()
-            for tag in sorted(agent.used_citations, key=agent._citation_tag_number):
-                ref = str((agent.citation_map.get(tag) or {}).get("ref") or "").strip()
-                if ref and ref not in seen_refs:
-                    references.append(f"- {ref}")
-                    seen_refs.add(ref)
-            if not references:
-                raise RuntimeError("No cited references remained after citation validation.")
-            full_draft_en = full_draft_en.rstrip() + "\n\n# References\n" + "\n\n".join(references)
-
-            pb.progress(94, text=tr(lang_code, "translation_running"))
-            update_heartbeat()
-            full_draft_translated = agent.translate_locked_manuscript(full_draft_en, target_lang)
-            agent.validate_translation_alignment(full_draft_en, full_draft_translated)
-
-            bj_time = datetime.now(timezone(timedelta(hours=8))).strftime("%Y%m%d_%H%M")
-            package = build_validated_package(
-                english_markdown=full_draft_en,
-                translated_markdown=full_draft_translated,
-                english_filename=f"Research_EN_{bj_time}.docx",
-                translated_filename=f"Research_{output_suffix}_{bj_time}.docx",
-                data_metadata=data_metadata,
-                mode=st.session_state.run_mode,
-                citation_audit=agent.citation_audit_log,
-                literature_metrics=agent.get_source_metrics(),
-            )
-            generated_zip_bytes = package.zip_bytes
-            generated_zip_filename = f"SmartPaper_{bj_time}.zip"
-
-            # Persist the already validated package in the current user's task
-            # directory before asking Worker/D1 to commit quota. The download is
-            # exposed only after the browser receives completed+committed.
-            task_id = str((st.session_state.get("sp_generation_task") or {}).get("task_id") or "").strip()
-            if not task_id:
-                raise RuntimeError("package_task_id_missing")
-            pending_path = os.path.join(_task_output_dir(task_id), ".pending_" + generated_zip_filename)
-            with open(pending_path, "wb") as output_file:
-                output_file.write(generated_zip_bytes)
-            st.session_state["sp_pending_package"] = {
-                "pending_path": pending_path,
-                "filename": generated_zip_filename,
-            }
-            completion_finalization_attempted = True
-            _finish_managed_generation_task(
-                "completed",
-                current_node="package_validated",
-                delete_checkpoint=True,
-            )
-            pb.progress(100, text=tr(lang_code, "complete"))
-            render_monitor(agent, tr(lang_code, "complete"), "✅", tr(lang_code, "document_generated"))
-            
-    except Exception as e:
-        task_id = str((st.session_state.get("sp_generation_task") or {}).get("task_id") or "")
-        logging.exception(
-            "[SP_TASK] generation_exception task_id=%s node=%s error_type=%s",
-            task_id,
-            str(st.session_state.get("sp_current_node") or "generation")[:120],
-            type(e).__name__,
-        )
-        if not completion_finalization_attempted:
-            try:
-                _finish_managed_generation_task(
-                    "failed",
-                    safe_error_code="E_GENERATION_FAILED",
-                    safe_error_detail=str(e)[:800],
-                    current_node=str(st.session_state.get("sp_current_node") or "generation"),
-                    detach_checkpoint=True,
-                )
-            except Exception:
-                pass
-        st.error(tr(lang_code, "startup_error", error=e))
-    finally:
-        task_id = str((st.session_state.get("sp_generation_task") or {}).get("task_id") or "")
-        logging.info(
-            "[SP_TASK] pipeline_exit task_id=%s node=%s",
-            task_id,
-            str(st.session_state.get("sp_current_node") or "")[:120],
-        )
-        clear_heartbeat()
-        st.session_state.is_running = False
-        st.session_state.start_pipeline = False
-        st.session_state.resume_pipeline = False
-
-if st.session_state.zip_buffer:
-    st.success(tr(lang_code, "success"))
-    st.download_button(label=tr(lang_code, "download"), data=st.session_state.zip_buffer, file_name=st.session_state.zip_filename, mime="application/zip")
+            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)
+
+
+if getattr(st, "fragment", None) is not None:
+    @st.fragment(run_every=5)
+    def _task_monitor_fragment():
+        _render_task_status_once()
+    _task_monitor_fragment()
+else:
+    _render_task_status_once()

config.py

@@ -8,6 +8,9 @@ def load_config():
         "lit_per_page": int(os.environ.get("LIT_PER_PAGE", "100")),
         "lit_max_workers": int(os.environ.get("LIT_MAX_WORKERS", "5")),
         "chunk_words_default": int(os.environ.get("CHUNK_WORDS_DEFAULT", "1500")),
+        "target_words_min": 5000,
+        "target_words_default": 5000,
+        "target_words_max": 20000,
         "search_max_retries": max(1, int(os.environ.get("SEARCH_MAX_RETRIES", "4"))),
         "translation_batch_tokens": max(800, int(os.environ.get("TRANSLATION_BATCH_TOKENS", "2500"))),
         "translation_glossary_term_limit": max(10, int(os.environ.get("TRANSLATION_GLOSSARY_TERM_LIMIT", "40"))),
@@ -124,11 +127,15 @@ def load_config():
                 int(os.environ.get("CITATION_SEMANTIC_JSON_REPAIR_RETRIES", "1")),
             ),
         },
-        # Task/lease settings are consumed by the HF client. The Worker remains
-        # the authoritative source for task ownership and lock expiry.
-        "task_heartbeat_interval_seconds": max(
-            3,
-            int(os.environ.get("SMARTPAPER_TASK_HEARTBEAT_SECONDS", "5")),
+        # R3-Free task execution runs outside Streamlit. The executor renews
+        # the Worker lease only while real Python/model progress is recent.
+        "executor_heartbeat_interval_seconds": max(
+            10,
+            int(os.environ.get("SMARTPAPER_EXECUTOR_HEARTBEAT_SECONDS", "15")),
+        ),
+        "executor_activity_grace_seconds": max(
+            30,
+            int(os.environ.get("SMARTPAPER_EXECUTOR_ACTIVITY_GRACE_SECONDS", "45")),
         ),
         "manuscript_quality": {
             "abstract_max_words": int(os.environ.get("ABSTRACT_MAX_WORDS", "200")),
@@ -147,7 +154,7 @@ def load_config():
         
         # --- UI 多语言标签 ---
         "ui_labels": {
-            "中文": {"title": "📊 工商管理论文智能体", "btn_generate": "🚀 生成初稿", "btn_download": "📥 下载文件", "topic_label": "📝 论文名称", "keywords_label": "🔍 关键字", "words_label": "📏 目标总单词数 (12000-25000单词)", "success": "🎉 流水线推演完毕！双语文档已就绪。"},
+            "中文": {"title": "📊 工商管理论文智能体", "btn_generate": "🚀 生成初稿", "btn_download": "📥 下载文件", "topic_label": "📝 论文名称", "keywords_label": "🔍 关键字", "words_label": "📏 目标总单词数 (5000-20000单词)", "success": "🎉 流水线推演完毕！双语文档已就绪。"},
             "English": {"title": "📊 Business Administration Article Agent", "btn_generate": "🚀 Generate Draft", "btn_download": "📥 Download", "topic_label": "📝 Article Title", "keywords_label": "🔍 Keywords", "words_label": "📏 Target Word Count", "success": "🎉 Task complete!"},
             "Русский": {"title": "📊 Агент статей по бизнес-администрированию", "btn_generate": "🚀 Черновик", "btn_download": "📥 Скачать", "topic_label": "📝 Тема", "keywords_label": "🔍 Ключевые слова", "words_label": "📏 Количество слов", "success": "🎉 Задача выполнена!"},
             "Қазақша": {"title": "📊 Бизнес-басқару жөніндегі ғылыми мақала агенті", "btn_generate": "🚀 Жоба дайындау", "btn_download": "📥 Жүктеу", "topic_label": "📝 Мақала атауы", "keywords_label": "🔍 Түйін сөздер", "words_label": "📏 Мақсатты сөз саны", "success": "🎉 Тапсырма аяқталды!"}

core/generation_runner.py

@@ -0,0 +1,592 @@
+# -*- coding: utf-8 -*-
+"""R3-Free headless SmartPaper generation runner.
+
+The runner contains no Streamlit calls. It is safe to execute in a subprocess
+whose lifetime is independent from Streamlit ScriptRunner reruns and browser
+connections. Checkpoints are task-scoped, immutable-by-sequence JSON snapshots.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import math
+import os
+import re
+from datetime import datetime, timezone, timedelta
+from pathlib import Path
+from typing import Callable, Any
+
+from config import APP_CONFIG, AMJ_PROMPTS, ASQ_PROMPTS
+from core.meta_extractor import extract_golden_metadata
+from core.pipeline_agent import SSciPipelineAgent
+from core.production_guard import build_validated_package, replace_abstract
+from core.qualitative_engine import get_qualitative_engine
+from core.quantitative_engine import get_quantitative_engine
+from core.rag_engine import initialize_rag
+from utils.file_parser import extract_text_from_files
+
+ProgressCallback = Callable[..., None]
+
+
+def _atomic_json(path: Path, value: dict[str, Any]) -> 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 _clean_trailing_references(text, keywords=("reference", "references", "reference list", "参考文献")):
+    lines = str(text or "").split("\n")
+    for i in range(len(lines) - 1, max(-1, len(lines) - 100), -1):
+        if any(re.match(rf"^##?\s*{kw}", lines[i], re.IGNORECASE) for kw in keywords):
+            return "\n".join(lines[:i]).strip()
+    return str(text or "").strip()
+
+
+def _clean_markdown_headings(text):
+    return re.sub(r"(?m)^#+\s+(.*)$", r"**\1**", str(text or ""))
+
+
+def _canonical_request_hash(request_data: dict[str, Any]) -> str:
+    material = dict(request_data 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()
+
+
+class TaskCheckpointStore:
+    def __init__(self, task_dir: Path, request_hash: str, execution_epoch: int):
+        self.task_dir = Path(task_dir)
+        self.request_hash = str(request_hash)
+        self.execution_epoch = int(execution_epoch)
+        self.directory = self.task_dir / "checkpoints"
+        self.directory.mkdir(parents=True, exist_ok=True)
+        self.sequence = 0
+
+    def load_latest(self) -> dict[str, Any] | None:
+        candidates: list[tuple[int, Path, dict[str, Any]]] = []
+        for path in self.directory.glob("checkpoint_*.json"):
+            try:
+                payload = json.loads(path.read_text(encoding="utf-8"))
+                if not isinstance(payload, dict):
+                    continue
+                if payload.get("request_hash") != self.request_hash:
+                    continue
+                seq = int(payload.get("checkpoint_seq") or 0)
+                if seq <= 0:
+                    continue
+                candidates.append((seq, path, payload))
+            except Exception:
+                continue
+        if not candidates:
+            return None
+        seq, _, payload = max(candidates, key=lambda item: item[0])
+        self.sequence = seq
+        return payload
+
+    def save(self, stage: str, state: dict[str, Any]) -> tuple[int, str]:
+        self.sequence += 1
+        stage_name = re.sub(r"[^A-Za-z0-9_.-]", "_", str(stage or "checkpoint"))[:80]
+        payload = dict(state or {})
+        payload.update(
+            {
+                "request_hash": self.request_hash,
+                "checkpoint_seq": self.sequence,
+                "checkpoint_stage": stage_name,
+                "checkpoint_execution_epoch": self.execution_epoch,
+                "checkpoint_saved_at": datetime.now(timezone.utc).isoformat(),
+            }
+        )
+        path = self.directory / f"checkpoint_{self.sequence:05d}_{stage_name}.json"
+        _atomic_json(path, payload)
+        latest = self.directory / "latest.json"
+        _atomic_json(latest, payload)
+        return self.sequence, stage_name
+
+
+class GenerationRunner:
+    def __init__(
+        self,
+        *,
+        task_dir: str | os.PathLike,
+        request_data: dict[str, Any],
+        execution_epoch: int,
+        progress_callback: ProgressCallback | None = None,
+        cancel_checker: Callable[[], bool] | None = None,
+    ):
+        self.task_dir = Path(task_dir)
+        self.request = dict(request_data)
+        self.execution_epoch = int(execution_epoch)
+        self.progress_callback = progress_callback or (lambda **_: None)
+        self.cancel_checker = cancel_checker or (lambda: False)
+        expected_hash = str(self.request.get("request_hash") or "")
+        actual_hash = _canonical_request_hash(self.request)
+        if not expected_hash or expected_hash != actual_hash:
+            raise RuntimeError("request_hash_mismatch")
+        self.request_hash = expected_hash
+        self.checkpoints = TaskCheckpointStore(self.task_dir, self.request_hash, self.execution_epoch)
+        self.output_dir = self.task_dir / "outputs"
+        self.output_dir.mkdir(parents=True, exist_ok=True)
+        self.agent = None
+
+    def _progress(self, pct: int, node: str, *, checkpoint_seq=None, checkpoint_stage=None) -> None:
+        agent = self.agent
+        self.progress_callback(
+            progress_pct=max(0, min(100, int(pct))),
+            current_node=str(node or "")[:240],
+            checkpoint_seq=checkpoint_seq,
+            checkpoint_stage=checkpoint_stage,
+            ds_calls=int(getattr(agent, "ds_calls", 0) or 0),
+            s2_calls=int(getattr(agent, "s2_calls", 0) or 0),
+            oa_calls=int(getattr(agent, "oa_calls", 0) or 0),
+        )
+
+    def _checkpoint(self, stage: str, state: dict[str, Any], pct: int) -> None:
+        seq, normalized_stage = self.checkpoints.save(stage, state)
+        self._progress(pct, normalized_stage, checkpoint_seq=seq, checkpoint_stage=normalized_stage)
+
+    @staticmethod
+    def _stream_text(generator) -> str:
+        return "".join(str(chunk or "") for chunk in generator)
+
+    @staticmethod
+    def _bounded_target(base_target: int, accumulated_deviation: float, remaining_including_current: int, safe_chunk: int) -> int:
+        base = max(80, int(base_target))
+        correction = -float(accumulated_deviation) / max(1, int(remaining_including_current))
+        lower = max(80, int(round(base * 0.75)))
+        upper = min(int(safe_chunk * 1.5), max(lower, int(round(base * 1.25))))
+        return max(lower, min(upper, int(round(base + correction))))
+
+    @staticmethod
+    def _english_count_for_target(manuscript: str) -> int:
+        body = re.split(r"(?m)^#\s+References\s*$", str(manuscript or ""), maxsplit=1)[0]
+        # Markdown punctuation is ignored; contractions/hyphenated terms stay one token.
+        return len(re.findall(r"\b[A-Za-z0-9]+(?:[-'][A-Za-z0-9]+)*\b", body))
+
+    def _validate_target_length(self, manuscript: str, target_words: int) -> None:
+        quality = APP_CONFIG.get("manuscript_validation", {})
+        minimum = int(target_words * float(quality.get("minimum_total_ratio", 0.85)))
+        maximum = int(target_words * float(quality.get("maximum_total_ratio", 1.15)))
+        actual = self._english_count_for_target(manuscript)
+        if actual < minimum or actual > maximum:
+            raise RuntimeError(
+                f"English manuscript length outside validated range: actual={actual}, "
+                f"required={minimum}-{maximum}, target={target_words}."
+            )
+
+    def _resolve_paths(self, items) -> list[str]:
+        output = []
+        for item in items or []:
+            path = (self.task_dir / str(item)).resolve()
+            if self.task_dir.resolve() not in path.parents:
+                raise RuntimeError("input_path_escape")
+            if path.is_file():
+                output.append(str(path))
+        return output
+
+    def run(self) -> dict[str, Any]:
+        req = self.request
+        mode = str(req.get("mode") or "").lower()
+        if mode not in {"amj", "asq"}:
+            raise RuntimeError("invalid_generation_mode")
+        target_words = int(req.get("target_words") or 0)
+        if not APP_CONFIG["target_words_min"] <= target_words <= APP_CONFIG["target_words_max"]:
+            raise RuntimeError("target_words_out_of_range")
+
+        ds_key = str(os.environ.get("DEEPSEEK_API_KEY") or "").strip()
+        s2_key = str(os.environ.get("S2_API_KEY") or "").strip()
+        if not ds_key:
+            raise RuntimeError("DEEPSEEK_API_KEY_missing")
+
+        if mode == "amj":
+            APP_CONFIG.update(AMJ_PROMPTS)
+        else:
+            APP_CONFIG.update(ASQ_PROMPTS)
+
+        agent = SSciPipelineAgent(ds_key, s2_key)
+        self.agent = agent
+        agent.set_cancel_checker(self.cancel_checker)
+        # DeepSeek/search activity marks the executor as genuinely alive; the
+        # heartbeat thread decides separately when to renew D1's lease.
+        agent.set_execution_progress_hook(lambda: self._progress_activity())
+        safe_chunk_words = int(agent.chunk_words_default)
+        quant_engine = get_quantitative_engine()
+        qual_engine = get_qualitative_engine(agent)
+
+        state = self.checkpoints.load_latest() or {}
+        if state:
+            logging.info(
+                "[SP_EXECUTOR] resume checkpoint seq=%s stage=%s",
+                state.get("checkpoint_seq"), state.get("checkpoint_stage"),
+            )
+
+        topic = str(req.get("topic") or "").strip()
+        keywords = str(req.get("keywords") or "").strip()
+        target_lang = str(req.get("target_lang") or "").strip()
+        output_suffix = str(req.get("output_suffix") or "Target").strip()
+        interface_language = str(req.get("interface_language") or "zh").strip()
+        reference_paths = self._resolve_paths(req.get("reference_files") or [])
+        data_paths = self._resolve_paths(req.get("data_files") or [])
+        interview_paths = self._resolve_paths(req.get("interview_files") or [])
+
+        # Restore agent state before continuing any dependent step.
+        if isinstance(state.get("agent_state"), dict):
+            agent.load_state(state["agent_state"])
+
+        user_lit_context = str(state.get("user_lit_context") or "")
+        user_data_md = str(state.get("user_data_md") or "")
+        data_metadata = dict(state.get("data_metadata") or {})
+
+        if not state.get("evidence_ready"):
+            self._progress(3, "inputs_and_evidence")
+            if reference_paths:
+                user_lit_context = extract_text_from_files(reference_paths[:20])
+                all_paper_texts = [extract_text_from_files([path]) for path in reference_paths[:20]]
+                agent.global_memory["core_golden_metadata"] = extract_golden_metadata(agent, all_paper_texts)
+            if mode == "amj":
+                data_file = data_paths[0] if data_paths else None
+                _, user_data_md, _, data_metadata = quant_engine.resolve_empirical_data(data_file, interface_language)
+            else:
+                raw_interviews = extract_text_from_files(interview_paths[:1]) if interview_paths else ""
+                user_data_md, _, data_metadata = qual_engine.resolve_qualitative_data(topic, keywords, raw_interviews)
+            state.update(
+                {
+                    "topic": topic,
+                    "keywords": keywords,
+                    "target_lang": target_lang,
+                    "target_words": target_words,
+                    "mode": mode,
+                    "interface_language": interface_language,
+                    "user_lit_context": user_lit_context,
+                    "user_data_md": user_data_md,
+                    "data_metadata": data_metadata,
+                    "agent_state": agent.export_state(),
+                    "evidence_ready": True,
+                }
+            )
+            self._checkpoint("evidence_ready", state, 6)
+
+        refs_text = str(state.get("refs_text") or "")
+        if not refs_text:
+            self._progress(8, "S2 + OpenAlex")
+            refs_text = agent.fetch_references_parallel(
+                topic,
+                keywords,
+                min_required=50,
+                progress_callback=lambda: self._progress(9, "S2 + OpenAlex"),
+            )
+            if len(agent.citation_map) < 50:
+                raise RuntimeError(f"Insufficient eligible references: {len(agent.citation_map)}/50")
+            if user_lit_context:
+                core_refs = agent.parse_user_core_literature(user_lit_context)
+                if core_refs:
+                    refs_text = (
+                        "【USER PROVIDED CORE LITERATURE】\n" + core_refs
+                        + "\n\n【RETRIEVED LITERATURE】\n" + refs_text
+                    )
+            state.update({"refs_text": refs_text, "agent_state": agent.export_state()})
+            self._checkpoint("literature_ready", state, 11)
+
+        rag_engine = initialize_rag([value for value in (user_lit_context, refs_text) if value])
+
+        synthesis_result = str(state.get("synthesis_result") or "")
+        if not synthesis_result:
+            self._progress(12, "literature_synthesis")
+            synthesis_result = self._stream_text(
+                agent.stream_synthesize_literature(topic, refs_text, user_core_literature=user_lit_context)
+            )
+            synthesis_result = re.sub(r"<think>.*?</think>", "", synthesis_result, flags=re.DOTALL).strip()
+            state.update({"synthesis_result": synthesis_result, "agent_state": agent.export_state()})
+            self._checkpoint("synthesis_ready", state, 13)
+
+        adaptive_structure = state.get("adaptive_structure")
+        if not isinstance(adaptive_structure, list) or not adaptive_structure:
+            self._progress(14, "anchors_and_outline")
+            anchor_response = self._stream_text(
+                agent.stream_generate_anchors(
+                    topic, keywords, synthesis_result, refs_text,
+                    data_evidence=user_data_md, mode=mode,
+                )
+            )
+            anchor_clean = re.sub(r"<think>.*?</think>", "", anchor_response, flags=re.DOTALL).strip()
+            abstract_match = re.search(
+                r"-+\s*ABSTRACT\s*-+\s*(.*?)(?=\s*-+\s*CONCLUSION\s*-+|$)",
+                anchor_clean, re.DOTALL | re.IGNORECASE,
+            )
+            conclusion_match = re.search(
+                r"-+\s*CONCLUSION\s*-+\s*(.*?)$", anchor_clean, re.DOTALL | re.IGNORECASE,
+            )
+            abstract_anchor = abstract_match.group(1).strip() if abstract_match else anchor_clean
+            conclusion_anchor = conclusion_match.group(1).strip() if conclusion_match else ""
+            agent.global_memory["abstract_anchor"] = abstract_anchor
+            agent.global_memory["conclusion_anchor"] = conclusion_anchor
+
+            plan = agent.generate_outline(topic, keywords, mode=mode, data_metadata=data_metadata)
+            en_topic = str(plan.get("en_topic", topic))
+            structure = plan.get("structure", [])
+            en_abstract_clean = _clean_markdown_headings(
+                re.sub(r"(?i)^abstract[:\s]*", "", str(abstract_anchor)).strip()
+            )
+            full_draft_en = f"# {en_topic}\n\n## Abstract\n{en_abstract_clean}\n\n## Keywords\n{keywords}\n\n"
+            adaptive_structure = agent.build_generation_tasks(structure, target_words, safe_chunk_words)
+            for task in adaptive_structure:
+                task["is_first"] = bool(task.get("section_first"))
+            jit_texts = agent.prefetch_jit_literature(topic, keywords, adaptive_structure)
+            if jit_texts:
+                rag_engine._initialize_chunks([jit_texts])
+            state.update(
+                {
+                    "adaptive_structure": adaptive_structure,
+                    "full_draft_en": full_draft_en,
+                    "full_draft_translated": "",
+                    "completed_sections": [],
+                    "context_tail": "",
+                    "accumulated_deviation": 0.0,
+                    "agent_state": agent.export_state(),
+                    "outline_ready": True,
+                }
+            )
+            self._checkpoint("outline_ready", state, 15)
+        else:
+            full_draft_en = str(state.get("full_draft_en") or "")
+            # JIT text itself is already represented in agent citation state;
+            # refs + user literature rebuild a safe RAG baseline on resume.
+
+        completed_sections = list(state.get("completed_sections") or [])
+        context_tail = str(state.get("context_tail") or "")
+        accumulated_deviation = float(state.get("accumulated_deviation") or 0.0)
+        all_cite_keys = list(agent.citation_map.keys())
+        valid_chunks = [
+            item for item in adaptive_structure
+            if not any(word in item["section"].lower() for word in ["abstract", "keyword", "appendix"])
+        ]
+        allocation_map: dict[str, list[str]] = {}
+        current_cite_idx = 0
+        for idx, item in enumerate(valid_chunks):
+            count = max(1, math.ceil(len(all_cite_keys) / max(1, len(valid_chunks))))
+            allocation_map[item["display_name"]] = (
+                all_cite_keys[current_cite_idx:]
+                if idx == len(valid_chunks) - 1
+                else all_cite_keys[current_cite_idx:current_cite_idx + count]
+            )
+            current_cite_idx += count
+
+        total_steps = len(adaptive_structure)
+        for sec in adaptive_structure:
+            display_name = sec["display_name"]
+            if display_name in completed_sections:
+                continue
+            if self.cancel_checker():
+                raise RuntimeError("task_cancel_requested")
+            remaining = sum(1 for item in adaptive_structure if item["display_name"] not in completed_sections)
+            adjusted_target = self._bounded_target(
+                int(sec["target_words"]), accumulated_deviation, remaining, safe_chunk_words
+            )
+            sec_title = sec["section"]
+            subsection = sec.get("subsection")
+            is_quarantined = any(
+                word in sec_title.lower()
+                for word in ["abstract", "keyword", "conclusion", "references", "appendix"]
+            )
+            assigned_tags = ", ".join(allocation_map.get(display_name, [])) or "the single most relevant source"
+            citation_rule = (
+                "NO IN-TEXT CITATIONS ALLOWED IN THIS SECTION."
+                if is_quarantined
+                else "Use claim-specific citation. Each sentence may contain at most one source tag. "
+                     f"Candidate tags: {assigned_tags}. <AVAILABLE_CITATIONS>{refs_text}</AVAILABLE_CITATIONS>"
+            )
+            memory_context = agent._build_memory_context(sec_title, 2500)
+            memory_block = f"\n<GLOBAL_MEMORY>\n{memory_context}\n</GLOBAL_MEMORY>\n" if memory_context else ""
+            snippets = rag_engine.get_relevant_context(display_name) if rag_engine is not None else ""
+            if snippets:
+                memory_block += f"\n<TARGETED_RAG>\n{snippets}\n</TARGETED_RAG>\n"
+            heading_directive = "Do not output the section heading; it will be inserted by the system."
+            if subsection:
+                heading_directive += (
+                    f" Do not output the subsection heading '{subsection}'; it will also be inserted by the system. "
+                    "Use level-3 headings only when analytically necessary."
+                )
+            elif sec_title.lower() in {"introduction", "conclusion"}:
+                heading_directive += " Do not use additional headings."
+            else:
+                heading_directive += (
+                    " Use level-2 or level-3 headings only when they clarify the argument; "
+                    "never use level-4 headings or numbered headings."
+                )
+            table_rule = (
+                "Present statistical evidence in standard Markdown tables with plain-text captions."
+                if sec["is_method_or_result"] and mode == "amj" else ""
+            )
+            concepts = ", ".join(sec.get("concepts") or [])
+            qualitative_rule = (
+                f"Assigned qualitative concepts: {concepts}. Use only their assigned representative quotations "
+                "and do not repeat a quotation already used elsewhere."
+                if concepts else ""
+            )
+            data_injection = (
+                f"\n<RESEARCH_EVIDENCE>\n{user_data_md}\n</RESEARCH_EVIDENCE>\n"
+                if user_data_md and sec["is_method_or_result"] else ""
+            )
+            completed_label = ", ".join(completed_sections)
+            anti_repeat = (
+                f"Already completed: {completed_label}. Do not restate their background, gap, or contribution."
+                if completed_label else ""
+            )
+            sys_prompt = APP_CONFIG.get("sys_academic_writer", "") + "\n" + APP_CONFIG.get("deep_refinement_directive", "")
+            user_prompt = f"""Write the assigned English manuscript unit.
+Section: {sec_title}
+Subsection: {subsection or 'None'}
+Target length: {int(adjusted_target * 0.9)}-{int(adjusted_target * 1.1)} words.
+{heading_directive}
+{citation_rule}
+{table_rule}
+{str(sec.get('required_content') or '')}
+{qualitative_rule}
+{anti_repeat}
+{memory_block}
+Literature synthesis: {synthesis_result}
+{data_injection}
+Previous tail: {context_tail[-2500:]}
+Output only final academic prose. Do not report the word count."""
+            if agent._count_tokens(sys_prompt) + agent._count_tokens(user_prompt) > 60000:
+                user_prompt = agent._truncate_context(user_prompt, 60000 - agent._count_tokens(sys_prompt))
+
+            current_index = len(completed_sections) + 1
+            pct = min(88, 15 + int((current_index / max(1, total_steps)) * 72))
+            self._progress(pct, display_name)
+            en_raw = self._stream_text(
+                agent.stream_deepseek(
+                    sys_prompt, user_prompt, max_tokens=7000,
+                    model_role="writing", thinking=False, allow_fallback=False,
+                )
+            )
+            en_raw = re.sub(r"<think>.*?</think>", "", en_raw, flags=re.S).strip()
+            en_raw = re.sub(rf"(?im)^#\s*{re.escape(sec_title)}\s*\n+", "", en_raw).strip()
+            if subsection:
+                en_raw = re.sub(rf"(?im)^##\s*{re.escape(subsection)}\s*\n+", "", en_raw).strip()
+            en_raw = re.sub(r"(?m)^(#{2,3})\s+[\d\.\s一二三四五六七八九十、]+\s*(.*)", r"\1 \2", en_raw)
+            en_raw = re.sub(r"(?m)^#{4,}\s+(.*)", r"**\1**", en_raw)
+            en_raw = re.sub(r"(?m)^#\s+(.+)$", r"## \1", en_raw)
+            if mode == "asq" and user_data_md and "appendix" not in sec_title.lower():
+                quote_dict = {tag: quote for quote, tag in re.findall(r'"([^"]+)"\s*\[(QUOTE-\d+)\]', user_data_md)}
+                for tag, quote in quote_dict.items():
+                    en_raw = en_raw.replace(tag, f'"{quote}"')
+            en_raw = agent.polish_academic_section(en_raw, display_name, abstract=False)
+            prefix = ""
+            if sec.get("section_first"):
+                prefix += f"# {sec_title}\n\n"
+            if sec.get("subsection_first") and subsection:
+                prefix += f"## {subsection}\n\n"
+            en_resolved = agent.audit_and_resolve_citations(
+                prefix + en_raw, section_name=display_name, quarantine=is_quarantined
+            )
+            full_draft_en += en_resolved + "\n\n"
+            agent._extract_memory_from_text(en_raw, sec_title)
+            context_tail = " ".join(en_raw.split()[-200:])
+            actual_words = len(en_raw.split())
+            accumulated_deviation += actual_words - adjusted_target
+            completed_sections.append(display_name)
+            state.update(
+                {
+                    "completed_sections": completed_sections,
+                    "full_draft_en": full_draft_en,
+                    "full_draft_translated": "",
+                    "context_tail": context_tail,
+                    "accumulated_deviation": accumulated_deviation,
+                    "agent_state": agent.export_state(),
+                    "data_metadata": data_metadata,
+                }
+            )
+            self._checkpoint(f"section_{len(completed_sections):03d}", state, pct)
+
+        if not state.get("english_finalized"):
+            self._progress(91, "english_validation")
+            full_draft_en = _clean_trailing_references(full_draft_en, ["Reference", "References", "Reference List"])
+            final_abstract = agent.generate_final_abstract(
+                topic, keywords, full_draft_en, data_evidence=user_data_md, mode=mode
+            )
+            full_draft_en = replace_abstract(full_draft_en, final_abstract)
+            agent.validate_manuscript_structure(full_draft_en, adaptive_structure, translated=False)
+            if mode == "asq":
+                repeated_quotes = agent.find_repeated_qualitative_quotes(full_draft_en, data_metadata)
+                if repeated_quotes:
+                    raise RuntimeError(f"Repeated qualitative quotations: {repeated_quotes[:5]}")
+            references = []
+            seen_refs = set()
+            for tag in sorted(agent.used_citations, key=agent._citation_tag_number):
+                ref = str((agent.citation_map.get(tag) or {}).get("ref") or "").strip()
+                if ref and ref not in seen_refs:
+                    references.append(f"- {ref}")
+                    seen_refs.add(ref)
+            if not references:
+                raise RuntimeError("No cited references remained after citation validation.")
+            full_draft_en = full_draft_en.rstrip() + "\n\n# References\n" + "\n\n".join(references)
+            self._validate_target_length(full_draft_en, target_words)
+            state.update(
+                {
+                    "full_draft_en": full_draft_en,
+                    "english_finalized": True,
+                    "agent_state": agent.export_state(),
+                }
+            )
+            self._checkpoint("english_complete", state, 93)
+        else:
+            full_draft_en = str(state.get("full_draft_en") or "")
+
+        full_draft_translated = str(state.get("full_draft_translated") or "")
+        if not state.get("translation_complete") or not full_draft_translated:
+            self._progress(95, "translation")
+            full_draft_translated = agent.translate_locked_manuscript(full_draft_en, target_lang)
+            agent.validate_translation_alignment(full_draft_en, full_draft_translated)
+            state.update(
+                {
+                    "full_draft_translated": full_draft_translated,
+                    "translation_complete": True,
+                    "agent_state": agent.export_state(),
+                }
+            )
+            self._checkpoint("translation_complete", state, 96)
+
+        existing_output = str(state.get("output_filename") or "").strip()
+        if state.get("package_validated") and existing_output:
+            existing_path = self.output_dir / os.path.basename(existing_output)
+            if existing_path.is_file() and existing_path.stat().st_size > 0:
+                self._progress(99, "package_validated")
+                return {"output_filename": existing_path.name, "output_path": str(existing_path)}
+
+        self._progress(97, "packaging")
+        bj_time = datetime.now(timezone(timedelta(hours=8))).strftime("%Y%m%d_%H%M")
+        package = build_validated_package(
+            english_markdown=full_draft_en,
+            translated_markdown=full_draft_translated,
+            english_filename=f"Research_EN_{bj_time}.docx",
+            translated_filename=f"Research_{output_suffix}_{bj_time}.docx",
+            data_metadata=data_metadata,
+            mode=mode,
+            citation_audit=agent.citation_audit_log,
+            literature_metrics=agent.get_source_metrics(),
+        )
+        output_filename = f"SmartPaper_{bj_time}.zip"
+        pending = self.output_dir / (".pending_" + output_filename)
+        pending.write_bytes(package.zip_bytes)
+        final = self.output_dir / output_filename
+        os.replace(pending, final)
+        state.update(
+            {
+                "output_filename": output_filename,
+                "package_validated": True,
+                "agent_state": agent.export_state(),
+            }
+        )
+        self._checkpoint("package_validated", state, 99)
+        return {"output_filename": output_filename, "output_path": str(final)}
+
+    def _progress_activity(self) -> None:
+        # This call intentionally does not update D1 directly. The executor's
+        # heartbeat thread uses it only to prove that real backend work is still
+        # occurring, which avoids browser-driven or timer-only zombie liveness.
+        try:
+            self.progress_callback(activity_only=True)
+        except TypeError:
+            pass

core/pipeline_agent.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 import os
-import streamlit as st
 import time
 import re
 import json
@@ -63,15 +62,11 @@ class SSciPipelineAgent:
                 "DEEPSEEK_API_KEY is empty. Configure it in Hugging Face "
                 "Space Settings > Variables and secrets."
             )
-        try: 
-            self.s2_key = st.secrets.get("S2_API_KEY", s2_key)
-        except: 
-            self.s2_key = os.environ.get("S2_API_KEY", s2_key)
-            
-        try: 
-            self.openalex_key = st.secrets.get("OA_API_KEY", "")
-        except: 
-            self.openalex_key = os.environ.get("OA_API_KEY", "")
+        # R3-Free: the generation engine is headless. Hugging Face Secrets are
+        # exposed to the process as environment variables; Streamlit is not a
+        # dependency of the executor or pipeline.
+        self.s2_key = str(os.environ.get("S2_API_KEY", s2_key) or "").strip()
+        self.openalex_key = str(os.environ.get("OA_API_KEY", "") or "").strip()
         self.openalex_email = str(APP_CONFIG.get("openalex_email") or "").strip()
             
         self.ds_calls = 0
@@ -2916,6 +2911,24 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
             raise DeepSeekRequestError(
                 f'Translation content coverage is too low: {ratio:.3f}.'
             )
+
+        # Russian and Kazakh both use Cyrillic, so a generic Cyrillic check is
+        # insufficient. Long Kazakh manuscripts must contain a meaningful
+        # number of Kazakh-specific letters; Russian manuscripts must not be
+        # dominated by those letters. This is deterministic and adds no API cost.
+        kazakh_specific = re.findall(r'[ӘәҒғҚқҢңӨөҰұҮүҺһІі]', target_body)
+        cyrillic = re.findall(r'[А-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі]', target_body)
+        if re.search(r'(?m)^#\s+Пайдаланылған әдебиеттер\s*$', target):
+            minimum_kazakh = max(8, int(len(cyrillic) * 0.003))
+            if len(kazakh_specific) < minimum_kazakh:
+                raise DeepSeekRequestError(
+                    'Kazakh language validation failed: insufficient Kazakh-specific letters.'
+                )
+        elif re.search(r'(?m)^#\s+Литература\s*$', target):
+            if cyrillic and len(kazakh_specific) / len(cyrillic) > 0.02:
+                raise DeepSeekRequestError(
+                    'Russian language validation failed: excessive Kazakh-specific letters.'
+                )
         return True
 
     def enforce_claim_specific_citations(self, text):

core/qualitative_engine.py

@@ -6,7 +6,6 @@ from __future__ import annotations
 import logging
 import re
 
-import streamlit as st
 
 from config import APP_CONFIG
 
@@ -69,35 +68,18 @@ Rules:
     def resolve_qualitative_data(self, topic, keywords, raw_interviews=""):
         has_user_data = bool(raw_interviews and len(raw_interviews) > 100)
         prompt = self._build_request(topic, keywords, raw_interviews)
-        model_label = self.agent.model_for("reasoning")
-
-        if has_user_data:
-            status_label = (
-                f"🧠 正在使用 {model_label} 分析访谈材料并冻结研究事实…"
-            )
-            complete_label = "✅ 访谈材料、研究事实与编码结构已完成"
-        else:
-            status_label = (
-                f"🧠 正在使用 {model_label} 构建一致的定性研究证据包…"
-            )
-            complete_label = "✅ 定性研究事实与编码结构已完成"
-
-        with st.status(status_label, expanded=True) as status:
-            placeholder = st.empty()
-            response = placeholder.write_stream(
-                self.agent.stream_deepseek(
-                    "You are a rigorous qualitative research analyst.",
-                    prompt,
-                    model_role="reasoning",
-                    thinking=True,
-                    allow_fallback=False,
-                )
-            )
-            status.update(
-                label=complete_label,
-                state="complete",
-                expanded=False,
+        # R3-Free engine code is headless. Streamlit rendering belongs to app.py;
+        # this method only produces deterministic evidence + metadata.
+        response = "".join(
+            str(chunk or "")
+            for chunk in self.agent.stream_deepseek(
+                "You are a rigorous qualitative research analyst.",
+                prompt,
+                model_role="reasoning",
+                thinking=True,
+                allow_fallback=False,
             )
+        )
 
         response = re.sub(
             r"<think>.*?</think>",

i18n.py

@@ -208,7 +208,7 @@ TEXTS: dict[str, dict[str, str]] = {
         "title": "📊 工商管理论文智能体",
         "topic_label": "📝 论文名称",
         "keywords_label": "🔍 关键字",
-        "words_label": "📏 目标总单词数（12000–25000单词）",
+        "words_label": "📏 目标总单词数（5000–20000单词）",
         "success": "🎉 流水线推演完毕！双语文档已就绪。",
         "download": "📥 下载文件",
         "system_busy_title": "⏳ 系统忙碌",
@@ -361,7 +361,7 @@ TEXTS: dict[str, dict[str, str]] = {
         "title": "📊 Business Administration Article Agent",
         "topic_label": "📝 Article Title",
         "keywords_label": "🔍 Keywords",
-        "words_label": "📏 Target Word Count (12,000–25,000 words)",
+        "words_label": "📏 Target Word Count (5,000–20,000 words)",
         "success": "🎉 The pipeline is complete. The bilingual documents are ready.",
         "download": "📥 Download Files",
         "system_busy_title": "⏳ System Busy",
@@ -514,7 +514,7 @@ TEXTS: dict[str, dict[str, str]] = {
         "title": "📊 Интеллектуальный агент для научных статей по управлению бизнесом",
         "topic_label": "📝 Название статьи",
         "keywords_label": "🔍 Ключевые слова",
-        "words_label": "📏 Целевой объём (12 000–25 000 слов)",
+        "words_label": "📏 Целевой объём (5 000–20 000 слов)",
         "success": "🎉 Работа конвейера завершена. Двуязычные документы готовы.",
         "download": "📥 Скачать файлы",
         "system_busy_title": "⏳ Система занята",
@@ -667,7 +667,7 @@ TEXTS: dict[str, dict[str, str]] = {
         "title": "📊 Бизнес-басқару жөніндегі ғылыми мақала интеллектуалды агенті",
         "topic_label": "📝 Мақала атауы",
         "keywords_label": "🔍 Түйін сөздер",
-        "words_label": "📏 Мақсатты көлем (12 000–25 000 сөз)",
+        "words_label": "📏 Мақсатты көлем (5 000–20 000 сөз)",
         "success": "🎉 Жұмыс үдерісі аяқталды. Екі тілдегі құжаттар дайын.",
         "download": "📥 Файлдарды жүктеу",
         "system_busy_title": "⏳ Жүйе бос емес",

managed_auth.py

@@ -29,39 +29,6 @@ from smartpaper_restore_bridge import edge_bridge
 EDGE_ASSERTION_COOKIE_NAME = "smartpaper_edge_assertion"
 EDGE_ASSERTION_AUDIENCE = "smartpaper-streamlit"
 ASSERTION_CLOCK_SKEW_SECONDS = 30
-EXECUTION_HEARTBEAT_PROOF_VERSION = "sp-execution-heartbeat-v1"
-
-
-def _execution_heartbeat_proof(task_id: str, current_node: str) -> tuple[int, str]:
-    """Create a short-lived HMAC proof that only the HF backend can mint."""
-    user = st.session_state.get("sp_user") or {}
-    numeric_id = user.get("numeric_id")
-    normalized_task = str(task_id or "").strip()
-    normalized_node = str(current_node or "")[:240]
-    if (
-        not isinstance(numeric_id, int)
-        or isinstance(numeric_id, bool)
-        or numeric_id <= 0
-        or not normalized_task
-    ):
-        raise RuntimeError("execution_heartbeat_identity_required")
-    user_id = str(numeric_id)
-    issued_at = int(time.time())
-    task_hash = hashlib.sha256(normalized_task.encode("utf-8")).hexdigest()
-    user_hash = hashlib.sha256(user_id.encode("utf-8")).hexdigest()
-    node_hash = hashlib.sha256(normalized_node.encode("utf-8")).hexdigest()
-    canonical = (
-        f"{EXECUTION_HEARTBEAT_PROOF_VERSION}:"
-        f"{task_hash}:{user_hash}:{node_hash}:{issued_at}"
-    )
-    signature = hmac.new(
-        _service_key().encode("utf-8"),
-        canonical.encode("ascii"),
-        hashlib.sha256,
-    ).hexdigest()
-    return issued_at, signature
-
-
 def _runtime_value(name: str) -> str:
     value = os.environ.get(name, "")
     if not value:
@@ -162,7 +129,7 @@ def _verify_edge_assertion(token: str) -> dict[str, Any]:
 
 def _clear_generation_session_state() -> None:
     for key in (
-        "sp_generation_task", "sp_current_node", "sp_last_task_heartbeat",
+        "sp_generation_task", "sp_current_node",
         "sp_generation_idempotency_key", "zip_buffer", "zip_filename",
         "is_running", "start_pipeline", "resume_pipeline", "run_mode",
         "rag_engine", "up_amj_ref", "up_amj_data", "up_asq_ref",
@@ -212,7 +179,6 @@ def _edge_call(
     command_id: str,
     payload: dict[str, Any] | None = None,
     *,
-    fire_and_forget: bool = False,
     key_prefix: str = "sp_edge",
 ) -> dict[str, Any] | None:
     return edge_bridge(
@@ -220,7 +186,6 @@ def _edge_call(
         command_id=command_id,
         api_origin=_api_base(),
         payload=payload or {},
-        fire_and_forget=fire_and_forget,
         key=_bridge_key(key_prefix, command_id),
     )
 
@@ -274,10 +239,18 @@ def get_account_snapshot(*, block: bool = False) -> dict[str, Any] | None:
     user = result.get("user")
     if isinstance(user, dict):
         st.session_state["sp_user"] = dict(user)
+    task = result.get("task")
+    latest_task = result.get("latest_task")
+    if isinstance(task, dict) and task.get("task_id"):
+        st.session_state["sp_generation_task"] = dict(task)
+    elif isinstance(latest_task, dict) and latest_task.get("task_id"):
+        st.session_state["sp_generation_task"] = dict(latest_task)
+    else:
+        st.session_state["sp_generation_task"] = None
     return result
 
 
-def start_generation_task(*, idempotency_key: str, mode: str, interface_language: str) -> dict[str, Any]:
+def start_generation_task(*, idempotency_key: str, mode: str, interface_language: str, target_words: int) -> dict[str, Any]:
     normalized = str(idempotency_key or "").strip()
     if not normalized:
         raise RuntimeError("idempotency_key_required")
@@ -288,6 +261,7 @@ def start_generation_task(*, idempotency_key: str, mode: str, interface_language
             "idempotency_key": normalized,
             "mode": str(mode or "").strip().lower(),
             "interface_language": str(interface_language or "zh").strip(),
+            "target_words": int(target_words),
         },
         key_prefix="sp_task_start",
     )
@@ -309,29 +283,6 @@ def recover_generation_task() -> dict[str, Any]:
     return dict(result)
 
 
-def heartbeat_generation_task(task_id: str, *, current_node: str = "") -> dict[str, Any]:
-    normalized = str(task_id or "").strip()
-    if not normalized:
-        raise RuntimeError("task_id_required")
-    normalized_node = str(current_node or "")[:240]
-    issued_at, proof = _execution_heartbeat_proof(normalized, normalized_node)
-    bucket = int(time.time() // 20)
-    _edge_call(
-        "task_heartbeat",
-        f"heartbeat:{normalized}:{bucket}",
-        {
-            "task_id": normalized,
-            "current_node": normalized_node,
-            "heartbeat_source": "execution",
-            "heartbeat_issued_at": issued_at,
-            "heartbeat_proof": proof,
-        },
-        fire_and_forget=True,
-        key_prefix="sp_task_heartbeat",
-    )
-    return {"ok": True, "queued": True, "cancel_requested": False}
-
-
 def generation_task_status(task_id: str) -> dict[str, Any]:
     normalized = str(task_id or "").strip()
     command_id = f"status:{normalized}:{int(time.time() // 5)}"
@@ -342,57 +293,10 @@ def generation_task_status(task_id: str) -> dict[str, Any]:
         st.stop()
     if not result.get("ok"):
         _raise_edge_error(result, "task_status_failed")
-    return dict(result)
-
-
-def finish_generation_task(
-    task_id: str,
-    *,
-    status: str,
-    safe_error_code: str = "",
-    safe_error_detail: str = "",
-    current_node: str = "",
-) -> dict[str, Any]:
-    """Finalize a task only after the browser receives Worker/D1 confirmation."""
-    normalized_task = str(task_id or "").strip()
-    normalized_status = str(status or "").strip().lower()
-    if not normalized_task:
-        raise RuntimeError("task_id_required")
-    if normalized_status not in {"completed", "failed", "cancelled", "resumable"}:
-        raise RuntimeError("invalid_finish_status")
-    command_id = f"finish:{normalized_task}:{normalized_status}"
-    result = _edge_call(
-        "task_finish",
-        command_id,
-        {
-            "task_id": normalized_task,
-            "status": normalized_status,
-            "safe_error_code": str(safe_error_code or "")[:80],
-            "safe_error_detail": str(safe_error_detail or "")[:800],
-            "current_node": str(current_node or "")[:240],
-        },
-        fire_and_forget=False,
-        key_prefix="sp_task_finish",
-    )
-    if result is None:
-        st.stop()
-    if not result.get("ok"):
-        _raise_edge_error(result, "task_finish_failed")
-    task = dict(result.get("task") or {})
-    reservation = dict(result.get("reservation") or {})
-    if str(task.get("task_id") or "") != normalized_task:
-        raise RuntimeError("task_finish_confirmation_mismatch")
-    if str(task.get("status") or "").lower() != normalized_status:
-        raise RuntimeError("task_finish_status_not_confirmed")
-    expected_reservation = {
-        "completed": "committed",
-        "failed": "released",
-        "cancelled": "released",
-        "resumable": "reserved",
-    }[normalized_status]
-    if str(reservation.get("status") or "").lower() != expected_reservation:
-        raise RuntimeError("task_quota_finalization_not_confirmed")
     _merge_quota_into_session(result)
+    task = result.get("task")
+    if isinstance(task, dict):
+        st.session_state["sp_generation_task"] = dict(task)
     return dict(result)
 
 
@@ -404,7 +308,6 @@ def cancel_generation_task(task_id: str | None = None) -> dict[str, Any]:
         "task_cancel",
         command_id,
         {"task_id": normalized} if normalized else {},
-        fire_and_forget=False,
         key_prefix="sp_task_cancel",
     )
     if result is None:

smartpaper_restore_bridge/__init__.py

@@ -41,7 +41,6 @@ def restore_bridge(
         expires_at=str(expires_at or ""),
         api_origin="",
         payload={},
-        fire_and_forget=False,
         height=0,
         default=None,
         key=key,
@@ -55,16 +54,13 @@ def edge_bridge(
     command_id: str,
     api_origin: str,
     payload: dict[str, Any] | None = None,
-    fire_and_forget: bool = False,
     key: str = "sp_edge_bridge",
 ) -> dict[str, Any] | None:
     """Execute one authenticated same-origin Worker command in the browser.
 
-    The first synchronous invocation usually returns ``None``. Streamlit reruns
-    after the browser completes the request and then returns the result. Only
-    non-terminal heartbeat calls may use ``fire_and_forget=True``. Task
-    completion, failure, cancellation, payment, and account mutations remain
-    synchronous so Python receives the Worker/D1 postcondition.
+    R3-Free keeps the browser bridge synchronous and limited to user-facing
+    operations. Execution liveness and terminal success/failure are owned by
+    the independent Python executor, never by this component.
     """
     value = _COMPONENT(
         action=str(action or ""),
@@ -73,7 +69,6 @@ def edge_bridge(
         expires_at="",
         api_origin=str(api_origin or ""),
         payload=dict(payload or {}),
-        fire_and_forget=bool(fire_and_forget),
         height=0,
         default=None,
         key=key,

smartpaper_restore_bridge/frontend/index.html

@@ -15,20 +15,14 @@
 
   const RESTORE_KEY = "smartpaper.restore_token.v1";
   const LEGACY_COOKIE_NAME = "__Host-smartpaper_restore";
-  const ACTIVE_TASK_KEY = "smartpaper.edge.active_task.v3";
-  const PENDING_OPS_KEY = "smartpaper.edge.pending_ops.v3";
   const RESULT_PREFIX = "smartpaper.edge.result.v3:";
   const MAX_TOKEN_LENGTH = 4096;
   const MAX_RESULT_AGE_MS = 10 * 60 * 1000;
-  const HEARTBEAT_MS = 25000;
   const REQUEST_TIMEOUT_MS = 20000;
   let lastResponseSignature = "";
-  let heartbeatTimer = null;
 
   const ACTION_PATHS = Object.freeze({
     task_start: "/edge/task/start",
-    task_heartbeat: "/edge/task/heartbeat",
-    task_finish: "/edge/task/finish",
     task_cancel: "/edge/task/cancel",
     task_status: "/edge/task/status",
     task_recover: "/edge/task/recover",
@@ -168,48 +162,6 @@
     } finally { window.clearTimeout(timer); }
   }
 
-  function pendingOperations() {
-    const value = readJson(PENDING_OPS_KEY, []);
-    return Array.isArray(value) ? value : [];
-  }
-
-  function storePending(items) { writeJson(PENDING_OPS_KEY, items.slice(-30)); }
-
-  function queuePending(operation) {
-    const items = pendingOperations();
-    if (!items.some(item => item.command_id === operation.command_id)) {
-      items.push(operation); storePending(items);
-    }
-  }
-
-  function removePending(commandId) {
-    storePending(pendingOperations().filter(item => item.command_id !== commandId));
-  }
-
-  async function drainPending() {
-    for (const operation of pendingOperations()) {
-      const result = await postJson(operation.api_origin, operation.path, operation.payload);
-      if (result.ok) {
-        removePending(operation.command_id);
-      } else if ([401,403].includes(Number(result.http_status || 0))) {
-        // Authentication may be restored by a new Alipay session. Keep the
-        // transient heartbeat operation out of the terminal task state.
-        removePending(operation.command_id);
-      }
-    }
-  }
-
-  function startBackgroundWork() {
-    // R2: the browser may retry commands that Python already emitted, but it
-    // must never manufacture task liveness merely because the tab remains open.
-    // Only real backend execution progress is allowed to enqueue task_heartbeat.
-    if (heartbeatTimer) return;
-    heartbeatTimer = window.setInterval(() => {
-      drainPending();
-    }, HEARTBEAT_MS);
-    drainPending();
-  }
-
   function cachedResult(commandId) {
     const cached = readJson(RESULT_PREFIX + commandId, null);
     if (!cached || Date.now() - Number(cached.saved_at || 0) > MAX_RESULT_AGE_MS) return null;
@@ -226,13 +178,6 @@
     const apiOrigin = normalizeOrigin(args.api_origin || "");
     if (!apiOrigin) return { ok:false, error:"invalid_api_origin", action, command_id:commandId };
     const payload = args.payload && typeof args.payload === "object" ? args.payload : {};
-    const fireAndForget = Boolean(args.fire_and_forget);
-
-    if (fireAndForget) {
-      queuePending({ command_id:commandId, api_origin:apiOrigin, path, payload });
-      drainPending();
-      return { ok:true, queued:true, action, command_id:commandId };
-    }
 
     const cached = cachedResult(commandId);
     if (cached) return cached;
@@ -241,12 +186,6 @@
     // be retryable on the next Streamlit rerun with the same idempotency key.
     if (result.ok) saveResult(commandId, result);
 
-    if (action === "task_start" && result.ok && result.task && result.task.task_id) {
-      writeJson(ACTIVE_TASK_KEY, { api_origin:apiOrigin, task_id:result.task.task_id, current_node:"generation" });
-    }
-    if (["task_finish", "task_cancel"].includes(action) && result.ok) {
-      window.localStorage.removeItem(ACTIVE_TASK_KEY);
-    }
     return result;
   }
 
@@ -272,7 +211,6 @@
   window.addEventListener("message", onRender);
   sendMessage("streamlit:componentReady", { apiVersion:1 });
   setFrameHeight();
-  startBackgroundWork();
 })();
 </script>
 </body>

task_executor.py

@@ -0,0 +1,391 @@
+# -*- coding: utf-8 -*-
+"""SmartPaper R3-Free independent task executor.
+
+Launched by Streamlit as a detached subprocess. It claims a D1 task with a
+server-to-server service key, runs the headless generation pipeline, renews the
+lease only while real execution activity is recent, and atomically asks the
+Worker to commit/release quota at terminal completion.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import os
+import re
+import sys
+import threading
+import time
+from pathlib import Path
+from urllib.parse import urlsplit
+
+import requests
+
+from config import APP_CONFIG
+from core.generation_runner import GenerationRunner
+from core.pipeline_agent import (
+    DeepSeekAPIError, DeepSeekAuthenticationError, DeepSeekBalanceError, DeepSeekRequestError,
+)
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    handlers=[logging.StreamHandler(sys.stdout)],
+)
+
+
+def _safe(value, limit=500):
+    text = str(value or "")
+    text = re.sub(r"(?i)(api[_ -]?key|authorization|bearer|token)\s*[:=]\s*[^\s,;]+", r"\1=[redacted]", text)
+    return text[:limit]
+
+
+def _api_origin() -> str:
+    value = str(os.environ.get("SMARTPAPER_AUTH_API") or "").strip().rstrip("/")
+    parsed = urlsplit(value)
+    if parsed.scheme != "https" or not parsed.netloc or parsed.path not in ("", "/") or parsed.query or parsed.fragment:
+        raise RuntimeError("SMARTPAPER_AUTH_API_invalid")
+    return f"https://{parsed.netloc}"
+
+
+def _service_key() -> str:
+    value = str(os.environ.get("SMARTPAPER_SERVICE_KEY") or "").strip()
+    if not value:
+        raise RuntimeError("SMARTPAPER_SERVICE_KEY_missing")
+    return value
+
+
+class ExecutorApiError(RuntimeError):
+    def __init__(self, error: str, status_code: int = 0, payload: dict | None = None):
+        super().__init__(str(error or "executor_api_error"))
+        self.error = str(error or "executor_api_error")
+        self.status_code = int(status_code or 0)
+        self.payload = dict(payload or {})
+
+
+def _is_recoverable_generation_failure(exc: Exception) -> bool:
+    """Return True only for transient execution failures that can use checkpoints.
+
+    Authentication/balance/request-validation failures are terminal and release
+    quota immediately. Transport/stream failures can be resumed from the last
+    durable task checkpoint without consuming a second quota unit.
+    """
+    if isinstance(exc, ExecutorApiError):
+        return exc.status_code == 429 or exc.status_code >= 500
+    if isinstance(exc, (DeepSeekAuthenticationError, DeepSeekBalanceError, DeepSeekRequestError)):
+        return False
+    if isinstance(exc, (requests.exceptions.Timeout, requests.exceptions.ConnectionError)):
+        return True
+    if isinstance(exc, DeepSeekAPIError):
+        return True
+    text = str(exc or "").lower()
+    transient_markers = (
+        "connection", "timed out", "timeout", "temporarily unavailable",
+        "streaming failed after", "api failed after", "stream was interrupted",
+        "insufficient eligible references",
+    )
+    return any(marker in text for marker in transient_markers)
+
+
+class ExecutorApi:
+    def __init__(self):
+        self.origin = _api_origin()
+        self.headers = {
+            "Authorization": f"Bearer {_service_key()}",
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+            "User-Agent": "SmartPaper-R3-Free-Executor/1.0",
+        }
+
+    def post(self, path: str, payload: dict, *, timeout=20) -> dict:
+        response = requests.post(
+            self.origin + path,
+            headers=self.headers,
+            json=payload,
+            timeout=timeout,
+        )
+        try:
+            data = response.json()
+        except Exception as exc:
+            raise RuntimeError(f"executor_worker_non_json_http_{response.status_code}") from exc
+        if not response.ok or data.get("ok") is False:
+            error = str(data.get("error") or data.get("reason") or f"HTTP_{response.status_code}")
+            raise ExecutorApiError(error, response.status_code, data)
+        return data
+
+    def claim(self, *, task_id: str, user_id: int, request_hash: str) -> dict:
+        return self.post(
+            "/executor/task/claim",
+            {"task_id": task_id, "user_id": user_id, "request_hash": request_hash},
+        )
+
+    def heartbeat(self, payload: dict) -> dict:
+        return self.post("/executor/task/heartbeat", payload)
+
+    def finish(self, payload: dict) -> dict:
+        return self.post("/executor/task/finish", payload, timeout=30)
+
+
+class ExecutionState:
+    def __init__(self, *, task_id: str, user_id: int, epoch: int, instance_id: str, api: ExecutorApi):
+        self.task_id = task_id
+        self.user_id = int(user_id)
+        self.epoch = int(epoch)
+        self.instance_id = str(instance_id)
+        self.api = api
+        self._lock = threading.Lock()
+        self._last_activity = time.monotonic()
+        self._progress_pct = 0
+        self._current_node = "executor_claimed"
+        self._checkpoint_seq = 0
+        self._checkpoint_stage = ""
+        self._ds_calls = 0
+        self._s2_calls = 0
+        self._oa_calls = 0
+        self._cancel = False
+        self._stop = threading.Event()
+        self._thread = None
+
+    def touch(self, *, progress_pct=None, current_node=None, checkpoint_seq=None, checkpoint_stage=None, ds_calls=None, s2_calls=None, oa_calls=None, activity_only=False):
+        with self._lock:
+            self._last_activity = time.monotonic()
+            if not activity_only:
+                if progress_pct is not None:
+                    self._progress_pct = max(self._progress_pct, max(0, min(100, int(progress_pct))))
+                if current_node:
+                    self._current_node = str(current_node)[:240]
+                if checkpoint_seq is not None and int(checkpoint_seq) >= self._checkpoint_seq:
+                    self._checkpoint_seq = int(checkpoint_seq)
+                    self._checkpoint_stage = str(checkpoint_stage or self._checkpoint_stage)[:120]
+                if ds_calls is not None:
+                    self._ds_calls = max(self._ds_calls, int(ds_calls))
+                if s2_calls is not None:
+                    self._s2_calls = max(self._s2_calls, int(s2_calls))
+                if oa_calls is not None:
+                    self._oa_calls = max(self._oa_calls, int(oa_calls))
+
+    def cancelled(self) -> bool:
+        with self._lock:
+            return bool(self._cancel)
+
+    def start_heartbeat(self):
+        self._thread = threading.Thread(target=self._heartbeat_loop, name="smartpaper-executor-heartbeat", daemon=True)
+        self._thread.start()
+
+    def stop_heartbeat(self):
+        self._stop.set()
+        if self._thread and self._thread.is_alive():
+            self._thread.join(timeout=2)
+
+    def _payload(self):
+        with self._lock:
+            return {
+                "task_id": self.task_id,
+                "user_id": self.user_id,
+                "execution_epoch": self.epoch,
+                "executor_instance_id": self.instance_id,
+                "current_node": self._current_node,
+                "progress_pct": self._progress_pct,
+                "checkpoint_seq": self._checkpoint_seq,
+                "checkpoint_stage": self._checkpoint_stage,
+                "ds_calls": self._ds_calls,
+                "s2_calls": self._s2_calls,
+                "oa_calls": self._oa_calls,
+            }, time.monotonic() - self._last_activity
+
+    def heartbeat_once(self):
+        payload, _ = self._payload()
+        result = self.api.heartbeat(payload)
+        if result.get("cancel_requested"):
+            with self._lock:
+                self._cancel = True
+        return result
+
+    def _heartbeat_loop(self):
+        interval = int(APP_CONFIG.get("executor_heartbeat_interval_seconds", 15))
+        grace = int(APP_CONFIG.get("executor_activity_grace_seconds", 45))
+        # Claim itself already set the first heartbeat; delay before the next one.
+        while not self._stop.wait(interval):
+            payload, idle_seconds = self._payload()
+            if idle_seconds > grace:
+                logging.warning(
+                    "[SP_EXECUTOR] heartbeat paused task_id=%s idle_s=%.1f node=%s",
+                    self.task_id, idle_seconds, payload.get("current_node"),
+                )
+                continue
+            try:
+                result = self.api.heartbeat(payload)
+                if result.get("cancel_requested"):
+                    with self._lock:
+                        self._cancel = True
+                    return
+            except ExecutorApiError as exc:
+                if exc.error in {
+                    "executor_superseded", "generation_lock_lost",
+                    "task_already_finalized", "task_cancel_requested",
+                } or exc.status_code in {404, 409, 410}:
+                    with self._lock:
+                        self._cancel = True
+                    logging.warning(
+                        "[SP_EXECUTOR] executor_no_longer_authoritative task_id=%s error=%s",
+                        self.task_id, _safe(exc.error, 180),
+                    )
+                    return
+                logging.warning(
+                    "[SP_EXECUTOR] heartbeat_error task_id=%s error=%s",
+                    self.task_id, _safe(exc, 180),
+                )
+            except Exception as exc:
+                # A transport outage alone does not authorize another local
+                # writer. D1 lease/epoch ownership remains authoritative.
+                logging.warning(
+                    "[SP_EXECUTOR] heartbeat_error task_id=%s error=%s",
+                    self.task_id, _safe(exc, 180),
+                )
+
+
+def _load_request(task_dir: Path) -> dict:
+    path = task_dir / "request.json"
+    if not path.is_file():
+        raise RuntimeError("request_manifest_missing")
+    value = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(value, dict):
+        raise RuntimeError("request_manifest_invalid")
+    return value
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--task-dir", required=True)
+    parser.add_argument("--task-id", required=True)
+    parser.add_argument("--user-id", required=True, type=int)
+    args = parser.parse_args()
+
+    task_dir = Path(args.task_dir).resolve()
+    request_data = _load_request(task_dir)
+    if str(request_data.get("task_id") or "") != args.task_id:
+        raise RuntimeError("request_task_id_mismatch")
+    if int(request_data.get("user_id") or 0) != int(args.user_id):
+        raise RuntimeError("request_user_id_mismatch")
+    request_hash = str(request_data.get("request_hash") or "").strip()
+    if not request_hash:
+        raise RuntimeError("request_hash_missing")
+
+    api = ExecutorApi()
+    claim = api.claim(task_id=args.task_id, user_id=args.user_id, request_hash=request_hash)
+    task = dict(claim.get("task") or {})
+    execution = dict(claim.get("execution") or {})
+    epoch = int(execution.get("execution_epoch") or task.get("execution_epoch") or 0)
+    instance_id = str(execution.get("executor_instance_id") or "")
+    if epoch <= 0 or not instance_id:
+        raise RuntimeError("executor_claim_identity_missing")
+    if int(task.get("target_words") or 0) != int(request_data.get("target_words") or 0):
+        raise RuntimeError("target_words_claim_mismatch")
+
+    state = ExecutionState(
+        task_id=args.task_id,
+        user_id=args.user_id,
+        epoch=epoch,
+        instance_id=instance_id,
+        api=api,
+    )
+    state.start_heartbeat()
+    logging.info(
+        "[SP_EXECUTOR] claimed task_id=%s epoch=%s instance=%s",
+        args.task_id, epoch, instance_id[:12],
+    )
+
+    # If D1 says a prior checkpoint exists but the free Space's local disk no
+    # longer has it (container rebuild), do not pretend resume is possible.
+    remote_checkpoint_seq = int(task.get("checkpoint_seq") or 0)
+    local_checkpoints = list((task_dir / "checkpoints").glob("checkpoint_*.json")) if (task_dir / "checkpoints").is_dir() else []
+    if remote_checkpoint_seq > 0 and not local_checkpoints:
+        try:
+            api.finish(
+                {
+                    "task_id": args.task_id,
+                    "user_id": args.user_id,
+                    "execution_epoch": epoch,
+                    "executor_instance_id": instance_id,
+                    "status": "failed",
+                    "safe_error_code": "E_CHECKPOINT_MISSING",
+                    "safe_error_detail": "Checkpoint metadata exists but local checkpoint files are unavailable after runtime replacement; quota released.",
+                    "current_node": "checkpoint_missing",
+                }
+            )
+        finally:
+            state.stop_heartbeat()
+        return 2
+
+    try:
+        runner = GenerationRunner(
+            task_dir=task_dir,
+            request_data=request_data,
+            execution_epoch=epoch,
+            progress_callback=state.touch,
+            cancel_checker=state.cancelled,
+        )
+        result = runner.run()
+        state.touch(progress_pct=100, current_node="package_validated")
+        # One last accepted heartbeat publishes the final checkpoint metadata
+        # before the atomic terminal transaction.
+        try:
+            state.heartbeat_once()
+        except Exception as exc:
+            logging.warning("[SP_EXECUTOR] final_heartbeat_error task_id=%s error=%s", args.task_id, _safe(exc, 180))
+        if state.cancelled():
+            raise RuntimeError("task_cancel_requested")
+        api.finish(
+            {
+                "task_id": args.task_id,
+                "user_id": args.user_id,
+                "execution_epoch": epoch,
+                "executor_instance_id": instance_id,
+                "status": "completed",
+                "current_node": "package_validated",
+                "output_filename": str(result.get("output_filename") or "")[:240],
+            }
+        )
+        logging.info("[SP_EXECUTOR] completed task_id=%s epoch=%s", args.task_id, epoch)
+        return 0
+    except Exception as exc:
+        if state.cancelled():
+            logging.warning(
+                "[SP_EXECUTOR] stopped_non_authoritative task_id=%s epoch=%s",
+                args.task_id, epoch,
+            )
+            return 3
+        if _is_recoverable_generation_failure(exc):
+            # Do not terminally fail or release quota. Stopping the real-execution
+            # heartbeat lets the Worker move this epoch to `resumable`; the next
+            # epoch continues from the latest immutable local checkpoint.
+            logging.exception(
+                "[SP_EXECUTOR] recoverable_interruption task_id=%s epoch=%s error=%s",
+                args.task_id, epoch, _safe(exc, 240),
+            )
+            return 75
+        logging.exception("[SP_EXECUTOR] failed task_id=%s epoch=%s error=%s", args.task_id, epoch, _safe(exc, 240))
+        try:
+            api.finish(
+                {
+                    "task_id": args.task_id,
+                    "user_id": args.user_id,
+                    "execution_epoch": epoch,
+                    "executor_instance_id": instance_id,
+                    "status": "failed",
+                    "safe_error_code": "E_GENERATION_FAILED",
+                    "safe_error_detail": _safe(exc, 800),
+                    "current_node": "generation_failed",
+                }
+            )
+        except Exception as finish_exc:
+            # A newer epoch or user cancellation may already own/finalize the
+            # task. Never override it from a stale executor.
+            logging.warning("[SP_EXECUTOR] failure_finalize_rejected task_id=%s error=%s", args.task_id, _safe(finish_exc, 220))
+        return 1
+    finally:
+        state.stop_heartbeat()
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

utils/file_parser.py

@@ -1,5 +1,13 @@
 # -*- coding: utf-8 -*-
-import streamlit as st
+"""Headless file parsing helpers used by both Streamlit and R3-Free executor."""
+from __future__ import annotations
+
+import io
+import logging
+import os
+from pathlib import Path
+from typing import BinaryIO, Iterable
+
 from docx import Document
 from config import APP_CONFIG
 
@@ -10,65 +18,114 @@ except ImportError:
     PANDAS_AVAILABLE = False
 
 try:
-    import fitz  # PyMuPDF 的导入名称是 fitz
+    import pymupdf
     PYMUPDF_AVAILABLE = True
 except ImportError:
     PYMUPDF_AVAILABLE = False
 
-def extract_text_from_files(uploaded_files, max_chars=None):
+
+def _name(file) -> str:
+    if isinstance(file, (str, os.PathLike, Path)):
+        return os.fspath(file)
+    return str(getattr(file, "name", "uploaded_file"))
+
+
+def _read_bytes(file) -> bytes:
+    if isinstance(file, (str, os.PathLike, Path)):
+        return Path(file).read_bytes()
+    try:
+        file.seek(0)
+    except Exception:
+        pass
+    if hasattr(file, "getvalue"):
+        data = file.getvalue()
+    else:
+        data = file.read()
+    try:
+        file.seek(0)
+    except Exception:
+        pass
+    return bytes(data or b"")
+
+
+def _binary_stream(file) -> io.BytesIO | BinaryIO:
+    if isinstance(file, (str, os.PathLike, Path)):
+        return io.BytesIO(Path(file).read_bytes())
+    try:
+        file.seek(0)
+    except Exception:
+        pass
+    return file
+
+
+def extract_text_from_files(uploaded_files: Iterable, max_chars=None):
     if max_chars is None:
         max_chars = APP_CONFIG.get("max_extract_chars", 150000)
-        
+
     extracted_text = ""
-    for file in uploaded_files:
+    for file in uploaded_files or []:
+        filename = _name(file)
+        lower_name = filename.lower()
         try:
-            file.seek(0)
-            if file.name.lower().endswith('.pdf') and PYMUPDF_AVAILABLE:
-                # 从内存中读取文件字节流
-                pdf_bytes = file.read()
-                # 使用上下文管理器确保内存被及时释放
-                with fitz.open(stream=pdf_bytes, filetype="pdf") as pdf_doc:
+            if lower_name.endswith(".pdf") and PYMUPDF_AVAILABLE:
+                pdf_bytes = _read_bytes(file)
+                with pymupdf.open(stream=pdf_bytes, filetype="pdf") as pdf_doc:
                     for page in pdf_doc:
                         text = page.get_text()
-                        if text: 
+                        if text:
                             extracted_text += text + "\n"
-                        if len(extracted_text) > max_chars: 
+                        if len(extracted_text) > max_chars:
                             break
-            elif file.name.lower().endswith(('.docx', '.doc')):
-                doc = Document(file)
+            elif lower_name.endswith((".docx", ".doc")):
+                stream = _binary_stream(file)
+                doc = Document(stream)
                 for para in doc.paragraphs:
                     extracted_text += para.text + "\n"
-                    if len(extracted_text) > max_chars: 
+                    if len(extracted_text) > max_chars:
                         break
             else:
-                content = file.getvalue().decode('utf-8', errors='ignore')
-                extracted_text += content + "\n"
-            
+                extracted_text += _read_bytes(file).decode("utf-8", errors="ignore") + "\n"
+
             if len(extracted_text) > max_chars:
-                st.warning(f"⚠️ 文件 {file.name} 已安全截断至 {max_chars} 字符。")
+                logging.warning("File %s was truncated safely to %s characters.", filename, max_chars)
                 break
-        except Exception as e:
-            st.warning(f"⚠️ 解析失败: {e}")
+        except Exception as exc:
+            logging.warning("File parse failed for %s: %s", filename, exc)
         finally:
-            file.seek(0)
+            if not isinstance(file, (str, os.PathLike, Path)):
+                try:
+                    file.seek(0)
+                except Exception:
+                    pass
     return extracted_text[:max_chars]
 
+
 def extract_data_from_files(uploaded_file, max_chars=20000):
-    if not uploaded_file: 
+    if not uploaded_file:
         return ""
-        
+
+    filename = _name(uploaded_file)
+    lower_name = filename.lower()
     try:
-        uploaded_file.seek(0)
         if PANDAS_AVAILABLE:
-            df = pd.read_csv(uploaded_file) if uploaded_file.name.lower().endswith('.csv') else pd.read_excel(uploaded_file)
+            source = os.fspath(uploaded_file) if isinstance(uploaded_file, (str, os.PathLike, Path)) else uploaded_file
+            if not isinstance(uploaded_file, (str, os.PathLike, Path)):
+                try:
+                    uploaded_file.seek(0)
+                except Exception:
+                    pass
+            df = pd.read_csv(source) if lower_name.endswith(".csv") else pd.read_excel(source)
             if len(df) > 100:
-                st.warning(f"⚠️ 数据表超长，安全截取前100行。")
+                logging.warning("Data file %s truncated safely to first 100 rows.", filename)
                 df = df.head(100)
             return df.to_markdown(index=False)
-        else:
-            return uploaded_file.getvalue().decode('utf-8', errors='ignore')[:max_chars]
-    except Exception as e:
-        st.warning(f"⚠️ 解析失败: {e}")
+        return _read_bytes(uploaded_file).decode("utf-8", errors="ignore")[:max_chars]
+    except Exception as exc:
+        logging.warning("Data parse failed for %s: %s", filename, exc)
         return ""
     finally:
-        uploaded_file.seek(0)
+        if not isinstance(uploaded_file, (str, os.PathLike, Path)):
+            try:
+                uploaded_file.seek(0)
+            except Exception:
+                pass