Deploy SmartPaper production v1.1 R1Fix

.streamlit/desktop.ini

@@ -0,0 +1,2 @@
+[.ShellClassInfo]
+IconResource=C:\Program Files\Google\Drive File Stream\129.0.1.0\GoogleDriveFS.exe,27

HF_COMMIT_SHA.txt

@@ -0,0 +1 @@
+af60fe1bdcbab60f47f06586cca03210f7f61026

app.py

@@ -1,260 +1,52 @@
 # -*- 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
-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")
-
+"""SmartPaper R3-Free UI: Streamlit is an observer, never the paper executor."""
+import hashlib
+import json
+import logging
 import os
-import streamlit.components.v1 as components
-import time
 import re
-import io
-import zipfile
-import math
-import json
 import secrets
-from datetime import datetime, timezone, timedelta
-import pandas as pd
-import numpy as np
-import threading
-from concurrent.futures import ThreadPoolExecutor
-from streamlit.runtime.scriptrunner import add_script_run_ctx, get_script_run_ctx
-
-from config import APP_CONFIG, AMJ_PROMPTS, ASQ_PROMPTS, GENERIC_PROMPTS
-from utils.file_parser import extract_text_from_files, extract_data_from_files
-from utils.docx_builder import text_to_docx
-from core.pipeline_agent import SSciPipelineAgent
+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
 from managed_auth import (
+    cancel_generation_task,
+    generation_task_status,
+    recover_generation_task,
     require_alipay_login,
     render_logged_in_account,
+    render_payment_purchase,
     start_generation_task,
-    heartbeat_generation_task,
-    finish_generation_task,
 )
-from i18n import LANGUAGE_OPTIONS, get_amj_csv_template, get_asq_txt_template, get_output_suffix, get_reference_heading, 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"
+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
-OUTPUT_DIR = "outputs"
-os.makedirs(OUTPUT_DIR, exist_ok=True)
-
-META_SECTIONS_BLACKLIST = set(APP_CONFIG["blacklists"]["meta_sections"])
-
-# =====================================================================
-# [架构升级] 强硬的文件心跳锁机制，彻底解决跨设备/跨进程并发漏洞
-# =====================================================================
-HEARTBEAT_FILE = os.path.join(WORKSPACE_DIR, "heartbeat.lock")
-HEARTBEAT_TIMEOUT = 120  # 2分钟心跳超时
-
-def update_heartbeat():
-    """Refresh both the local file lock and the Cloudflare task lease."""
-    now = time.time()
-    try:
-        with open(HEARTBEAT_FILE, "w") as f:
-            f.write(str(now))
-    except Exception:
-        pass
-
-    task = st.session_state.get("sp_generation_task") or {}
-    task_id = str(task.get("task_id") or "").strip()
-    last_remote = float(st.session_state.get("sp_last_task_heartbeat") or 0.0)
-    if not task_id or now - last_remote < 25:
-        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:
-        # A transient heartbeat failure must not interrupt paper generation.
-        pass
-
-def is_system_busy():
-    if not os.path.exists(HEARTBEAT_FILE):
-        return False
-    try:
-        with open(HEARTBEAT_FILE, "r") as f:
-            last_active = float(f.read().strip())
-        if time.time() - last_active < HEARTBEAT_TIMEOUT:
-            return True
-    except:
-        pass
-    return False
-
-def clear_heartbeat():
-    try:
-        if os.path.exists(HEARTBEAT_FILE):
-            os.remove(HEARTBEAT_FILE)
-    except:
-        pass
-
-def stream_with_heartbeat(generator):
-    """包裹器：确保大模型长时间流式输出时，能够持续发送心跳，防止锁被释放"""
-    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)
 
-@st.cache_resource
-def get_global_lock(): 
-    return threading.Lock()
-
-global_lock = get_global_lock()
-
-def generate_psd_corr_matrix(n_constructs, max_corr=0.5):
-    if n_constructs == 1: return np.array([[1.0]])
-    base_corr = np.random.uniform(0.2, max_corr - 0.1) 
-    A = np.random.uniform(base_corr - 0.1, base_corr + 0.1, (n_constructs, n_constructs))
-    A = (A + A.T) / 2
-    np.fill_diagonal(A, 1.0)
-    min_eig = np.min(np.linalg.eigvals(A))
-    if min_eig < 0: A -= (min_eig - 0.01) * np.eye(n_constructs)
-    d = np.sqrt(np.diag(A))
-    return A / np.outer(d, d)
-
-def calculate_validity_metrics(df, constructs_info, true_loadings):
-    metrics = []
-    for const in constructs_info:
-        c_name = const['name']
-        items = const['items']
-        sub_df = df[items]
-        var_total = sub_df.sum(axis=1).var(ddof=1)
-        sum_var = sub_df.var(ddof=1).sum()
-        n_items = len(items)
-        alpha = 0.0 if var_total <= 0 else (n_items / (n_items - 1)) * (1 - sum_var / var_total)
-        c_loadings = np.array(true_loadings[c_name])
-        sum_loadings_sq = np.sum(c_loadings)**2
-        sum_sq_loadings = np.sum(c_loadings**2)
-        sum_error = np.sum(1 - c_loadings**2)
-        cr = sum_loadings_sq / (sum_loadings_sq + sum_error)
-        ave = sum_sq_loadings / len(c_loadings)
-        metrics.append({"Construct": c_name, "Items": n_items, "Cronbach's α": round(alpha, 3), "CR": round(cr, 3), "AVE": round(ave, 3)})
-    return pd.DataFrame(metrics)
-
-def generate_empirical_data_robust(construct_names=None, n=300, scale_points=7):
-    if not construct_names or len(construct_names) < 2: 
-        construct_names = ["Independent_Var", "Mediator_Var", "Dependent_Var"]
-        
-    n_constructs = len(construct_names)
-    corr = generate_psd_corr_matrix(n_constructs, max_corr=0.55)
-    latent_scores = np.random.multivariate_normal(mean=[0]*n_constructs, cov=corr, size=n)
-    data, true_loadings, constructs_info = {}, {}, []
-    target_mean = scale_points / 2.0 + 0.5
-    target_std = scale_points / 5.0
-    
-    for idx, c_name in enumerate(construct_names):
-        true_score = latent_scores[:, idx]
-        n_items = np.random.randint(3, 6) 
-        loadings = np.random.uniform(0.72, 0.88, n_items)
-        true_loadings[c_name] = loadings
-        item_names = []
-        for i, load in enumerate(loadings):
-            error = np.random.normal(0, np.sqrt(1 - load**2), n)
-            item_score = load * true_score + error 
-            raw_scaled = item_score * target_std + target_mean
-            item_int = np.clip(np.round(raw_scaled), 1, scale_points).astype(int)
-            col_name = f"{c_name}_{i+1}"
-            data[col_name] = item_int
-            item_names.append(col_name)
-        constructs_info.append({"name": c_name, "items": item_names})
-        
-    df = pd.DataFrame(data)
-    validity_df = calculate_validity_metrics(df, constructs_info, true_loadings)
-    mean_scores = pd.DataFrame({c['name']: df[c['items']].mean(axis=1) for c in constructs_info})
-    fl_matrix = mean_scores.corr().astype(object) 
-    
-    for i, row in validity_df.iterrows():
-        c_name = row['Construct']
-        sqrt_ave = np.sqrt(row['AVE'])
-        fl_matrix.loc[c_name, c_name] = f"({sqrt_ave:.3f})"
-        
-    for i in range(len(fl_matrix.columns)):
-        for j in range(i + 1, len(fl_matrix.columns)): 
-            fl_matrix.iloc[i, j] = ""
-            
-    desc_stats = mean_scores.mean().to_frame('Mean').join(mean_scores.std().to_frame('SD'))
-    return df, validity_df, desc_stats.join(fl_matrix)
-
-def resolve_empirical_data(uploaded_files=None, language="zh"):
-    if uploaded_files:
-        user_md = extract_data_from_files(uploaded_files)
-        if user_md:
-            return None, user_md, tr(language, "actual_stats_parsed")
-    
-    df, validity_df, fl_matrix_df = generate_empirical_data_robust()
-    simulated_md_prompt = f"""### Table 1: Reliability and Convergent Validity\n{validity_df.to_markdown(index=False)}\n### Table 2: Descriptive Statistics and Discriminant Validity (Fornell-Larcker Criterion)\n*Note: Diagonal elements in parentheses are the square roots of AVE.*\n{fl_matrix_df.to_markdown()}\n[STRICT EMPIRICAL INSTRUCTION]\n1. ZERO HALLUCINATION: You MUST report the exact metrics (Cronbach's α, CR, AVE, correlations) provided in the tables above.\n2. VALIDITY DEFENSE: Explicitly state that Convergent Validity is established because all AVEs > 0.5 and CRs > 0.7.\n3. DISCRIMINANT DEFENSE: Explicitly state that Discriminant Validity is satisfied per the Fornell-Larcker criterion, as the square root of AVE (diagonal values) exceed inter-construct correlations."""
-    status_msg = tr(language, "synthetic_stats_generated")
-    return df, simulated_md_prompt, status_msg
-
-def _translation_system_prompt(target_lang):
-    common = """Translate the supplied academic manuscript faithfully. Preserve every citation tag, number, percentage, coefficient, p-value, confidence interval, DOI, table structure, heading level, hypothesis label, variable abbreviation, and conclusion direction. Do not add, remove, strengthen, or weaken empirical claims. Use concise, accurate, and rigorous academic language. Do not use metaphors, analogies, rhetorical questions, promotional language, formulaic transitions, or mechanical repetition. Output only the translated manuscript."""
-    if target_lang == "Russian (Русский)":
-        return common + " Ensure idiomatic academic Russian, correct case government, gender-number agreement, verb aspect, punctuation, and terminology. Reorganize English syntax where necessary and avoid literal English calques. Keep journal titles, DOIs, and reference entries in their original form."
-    if target_lang == "Kazakh (Қазақша)":
-        return common + " Ensure idiomatic academic Kazakh, correct case and possessive suffixes, person and plural endings, vowel harmony, consonant assimilation, and natural Kazakh word order. Avoid Russian syntactic calques and unnecessary Russian loan forms. Use Kazakh-specific letters correctly: ә, ғ, қ, ң, ө, ұ, ү, һ, і. Keep journal titles, DOIs, and reference entries in their original form."
-    return common + " Use formal academic Chinese with clear concepts, explicit logic, and natural sentence structure."
-
-
-def async_translate_task(agent, en_text, placeholder, target_lang, quarantine_flag, interface_lang):
-    try:
-        sys_p = _translation_system_prompt(target_lang)
-        trans_text = agent.call_deepseek(sys_p, f"TEXT:\n{en_text}", max_tokens=6000, model_name="deepseek-chat")
-        trans_resolved = agent.resolve_citations(trans_text, quarantine=quarantine_flag)
-        placeholder.markdown(trans_resolved)
-        return trans_resolved
-    except Exception:
-        placeholder.error(tr(interface_lang, "translation_error"))
-        return ""
-
 
 def inject_file_uploader_i18n(language: str) -> None:
     """Localize Streamlit uploaders without modifying their React structure."""
@@ -462,1031 +254,504 @@ 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 = ""
+
+for key, default in {
+    "is_running": False,
+    "zip_buffer": None,
+    "zip_filename": "",
+    "run_mode": "amj",
+    "sp_generation_task": None,
+    "sp_pending_launch": None,
+    "sp_pending_resume_task_id": "",
+    "sp_pending_cancel_task_id": "",
+    "sp_payment_required": False,
+    "sp_terminal_notice": None,
+    "sp_task_recovery_initialized": False,
+    "sp_last_ui_task_status": "",
+}.items():
+    if key not in st.session_state:
+        st.session_state[key] = default
 
 
-def _safe_user_workspace_key(value):
+def _safe_key(value: str) -> str:
     cleaned = re.sub(r"[^A-Za-z0-9_.-]", "_", str(value or "").strip())
-    return cleaned[:80] or "anonymous"
+    return cleaned[:100] or "anonymous"
 
 
-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:
+def _new_generation_idempotency_key() -> str:
+    return f"gen_{int(time.time() * 1000)}_{secrets.token_urlsafe(18)}"
+
+
+def _atomic_json(path: Path, value: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
+    os.replace(temporary, path)
+
+
+def _canonical_request_hash(value: dict) -> str:
+    material = dict(value or {})
+    material.pop("request_hash", None)
+    encoded = json.dumps(material, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _task_dir(task_id: str) -> Path:
+    return Path(_user_workspace) / "tasks" / _safe_key(task_id)
+
+
+def _task_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
-    required = {
-        "topic", "keywords", "target_lang", "words",
-        "agent_state", "adaptive_structure", "completed_sections",
+    return _task_dir(task_id) / "outputs" / filename
+
+
+def _save_uploaded(items, inputs_dir: Path, prefix: str) -> list[str]:
+    saved = []
+    for index, item in enumerate(items or []):
+        if item is None:
+            continue
+        original = os.path.basename(str(getattr(item, "name", f"input_{index}")))
+        safe_name = re.sub(r"[^A-Za-z0-9_.-]", "_", original)[:140] or f"input_{index}"
+        filename = f"{prefix}_{index:02d}_{safe_name}"
+        path = inputs_dir / filename
+        data = item.getvalue() if hasattr(item, "getvalue") else item.read()
+        path.write_bytes(bytes(data or b""))
+        saved.append(str(Path("inputs") / filename))
+        try:
+            item.seek(0)
+        except Exception:
+            pass
+    return saved
+
+
+def _persist_task_request(task: dict, mode: str, lang_code: str) -> dict:
+    task_id = str(task.get("task_id") or "").strip()
+    numeric_id = (st.session_state.get("sp_user") or {}).get("numeric_id")
+    if not task_id or not isinstance(numeric_id, int) or numeric_id <= 0:
+        raise RuntimeError("task_identity_missing")
+    task_dir = _task_dir(task_id)
+    inputs_dir = task_dir / "inputs"
+    inputs_dir.mkdir(parents=True, exist_ok=True)
+
+    if mode == "amj":
+        refs = _save_uploaded(st.session_state.get("up_amj_ref") or [], inputs_dir, "ref")
+        data = _save_uploaded([st.session_state.get("up_amj_data")], inputs_dir, "data")
+        interviews = []
+    else:
+        refs = _save_uploaded(st.session_state.get("up_asq_ref") or [], inputs_dir, "ref")
+        data = []
+        interviews = _save_uploaded([st.session_state.get("up_asq_data")], inputs_dir, "interview")
+
+    request_data = {
+        "task_id": task_id,
+        "user_id": numeric_id,
+        "mode": mode,
+        "interface_language": lang_code,
+        "target_lang": get_target_language(lang_code),
+        "output_suffix": get_output_suffix(lang_code),
+        "topic": str(st.session_state.get("sp_topic") or DEFAULT_TOPIC),
+        "keywords": str(st.session_state.get("sp_keywords") or DEFAULT_KEYWORDS),
+        "target_words": int(st.session_state.get("sp_words") or APP_CONFIG["target_words_default"]),
+        "reference_files": refs,
+        "data_files": data,
+        "interview_files": interviews,
+        "created_at": datetime.now(timezone.utc).isoformat(),
     }
-    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
+    request_data["request_hash"] = _canonical_request_hash(request_data)
+    _atomic_json(task_dir / "request.json", request_data)
+    return request_data
 
 
-def _new_generation_idempotency_key() -> str:
-    return f"gen_{int(time.time() * 1000)}_{secrets.token_urlsafe(18)}"
+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 _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"] = ""
+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 _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()
+def _launch_executor(task_id: str, *, force: bool = False) -> int:
+    """Launch only after a PID-bound claim handshake is observed.
 
-    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()
+    ``subprocess.Popen`` proves only that the OS accepted a child process. R3.1
+    waits until that exact child has successfully claimed the D1 task and
+    written ``executor_claim.json``. A child that exits/import-fails before
+    claim is therefore never reported as a running executor.
+    """
+    task_dir = _task_dir(task_id)
+    request_path = task_dir / "request.json"
+    if not request_path.is_file():
+        raise RuntimeError("task_request_manifest_missing")
+    request_data = json.loads(request_path.read_text(encoding="utf-8"))
+    numeric_id = int(request_data.get("user_id") or 0)
+    request_hash = str(request_data.get("request_hash") or "").strip()
+    if numeric_id <= 0 or not request_hash:
+        raise RuntimeError("task_request_identity_invalid")
+
+    launch_marker_path = task_dir / "executor_launch.json"
+    claim_marker_path = task_dir / "executor_claim.json"
+    if launch_marker_path.is_file():
+        try:
+            marker = json.loads(launch_marker_path.read_text(encoding="utf-8"))
+            pid = int(marker.get("pid") or 0)
+            verified_executor = _pid_is_task_executor(pid, task_id, task_dir)
+            if verified_executor and not force:
+                claim_marker = {}
+                if claim_marker_path.is_file():
+                    try:
+                        claim_marker = json.loads(claim_marker_path.read_text(encoding="utf-8"))
+                    except Exception:
+                        claim_marker = {}
+                if int(claim_marker.get("pid") or 0) == pid and int(claim_marker.get("execution_epoch") or 0) > 0:
+                    return pid
+            if verified_executor and force:
+                try:
+                    os.kill(pid, signal.SIGTERM)
+                except OSError:
+                    pass
+        except Exception:
+            pass
 
-    response = start_generation_task(
-        idempotency_key=idempotency_key,
-        mode=mode,
-        interface_language=language,
+    # A stale claim file must never satisfy a new launch/resume handshake.
+    try:
+        claim_marker_path.unlink(missing_ok=True)
+    except TypeError:
+        if claim_marker_path.exists():
+            claim_marker_path.unlink()
+
+    log_path = task_dir / "executor.log"
+    task_dir.mkdir(parents=True, exist_ok=True)
+    command = [
+        sys.executable,
+        str(Path(__file__).resolve().parent / "task_executor.py"),
+        "--task-dir", str(task_dir.resolve()),
+        "--task-id", str(task_id),
+        "--user-id", str(numeric_id),
+    ]
+    executor_env = os.environ.copy()
+    executor_env["SMARTPAPER_EXECUTOR_LOG_PATH"] = str(log_path.resolve())
+    process = subprocess.Popen(
+        command,
+        cwd=str(Path(__file__).resolve().parent),
+        stdin=subprocess.DEVNULL,
+        # Inherit stdout/stderr so executor bootstrap/errors are visible in HF
+        # Container Logs; task_executor also writes the same log to executor.log.
+        close_fds=True,
+        start_new_session=True,
+        env=executor_env,
     )
-    task = dict(response.get("task") or {})
-    if not task.get("task_id"):
-        raise RuntimeError("task_start_missing_task_id")
+    _atomic_json(launch_marker_path, {"pid": process.pid, "launched_at": datetime.now(timezone.utc).isoformat()})
+    logging.info("[SP_UI] executor_spawned task_id=%s pid=%s", task_id, process.pid)
+
+    timeout_seconds = float(APP_CONFIG.get("executor_bootstrap_timeout_seconds", 20))
+    poll_seconds = max(0.05, float(APP_CONFIG.get("executor_bootstrap_poll_ms", 100)) / 1000.0)
+    deadline = time.monotonic() + timeout_seconds
+    while time.monotonic() < deadline:
+        if claim_marker_path.is_file():
+            try:
+                claim_marker = json.loads(claim_marker_path.read_text(encoding="utf-8"))
+            except Exception:
+                claim_marker = {}
+            if (
+                int(claim_marker.get("pid") or 0) == int(process.pid)
+                and str(claim_marker.get("task_id") or "") == str(task_id)
+                and str(claim_marker.get("request_hash") or "") == request_hash
+                and int(claim_marker.get("execution_epoch") or 0) > 0
+                and str(claim_marker.get("executor_instance_id") or "").strip()
+            ):
+                logging.info(
+                    "[SP_UI] executor_claim_confirmed task_id=%s pid=%s epoch=%s",
+                    task_id, process.pid, claim_marker.get("execution_epoch"),
+                )
+                return int(process.pid)
+        return_code = process.poll()
+        if return_code is not None:
+            logging.error("[SP_UI] executor_bootstrap_exit task_id=%s pid=%s rc=%s", task_id, process.pid, return_code)
+            raise RuntimeError(f"executor_bootstrap_failed_exit_{return_code}")
+        time.sleep(poll_seconds)
 
-    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()
-    return task, idempotency_key
+    try:
+        process.terminate()
+    except Exception:
+        pass
+    logging.error("[SP_UI] executor_claim_timeout task_id=%s pid=%s", task_id, process.pid)
+    raise RuntimeError("executor_claim_timeout")
 
 
-def _finish_managed_generation_task(
-    status: str,
-    *,
-    safe_error_code: str = "",
-    safe_error_detail: str = "",
-    current_node: str = "",
-):
-    task = st.session_state.get("sp_generation_task") or {}
+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()
+
+
+def _process_pending_resume() -> None:
+    task_id = str(st.session_state.get("sp_pending_resume_task_id") or "").strip()
+    if not task_id:
+        return
+    if not (_task_dir(task_id) / "request.json").is_file():
+        st.session_state["sp_pending_resume_task_id"] = ""
+        st.session_state["sp_pending_cancel_task_id"] = task_id
+        st.session_state["sp_terminal_notice"] = {
+            "kind": "error",
+            "message": "Local checkpoint/request files are unavailable after runtime replacement; the reserved quota will be released.",
+        }
+        st.rerun()
+    _launch_executor(task_id, force=True)
+    st.session_state["sp_pending_resume_task_id"] = ""
+    st.rerun()
+
+
+def _process_pending_launch(lang_code: str) -> None:
+    pending = st.session_state.get("sp_pending_launch")
+    if not isinstance(pending, dict):
+        return
+    mode = str(pending.get("mode") or "").strip().lower()
+    idempotency_key = str(pending.get("idempotency_key") or "").strip()
+    try:
+        response = start_generation_task(
+            idempotency_key=idempotency_key,
+            mode=mode,
+            interface_language=lang_code,
+            target_words=int(st.session_state.get("sp_words") or APP_CONFIG["target_words_default"]),
+        )
+    except Exception as exc:
+        st.session_state["sp_pending_launch"] = None
+        raw = str(exc or "")
+        if "payment_required" in raw:
+            st.session_state["sp_payment_required"] = True
+            st.error(tr(lang_code, "quota_exhausted"))
+            render_payment_purchase(lang_code)
+        else:
+            st.error(raw)
+        return
+    task = dict(response.get("task") or {})
     task_id = str(task.get("task_id") or "").strip()
     if not task_id:
-        return None
+        st.session_state["sp_pending_launch"] = None
+        st.error("task_start_missing_task_id")
+        return
+    st.session_state["sp_generation_task"] = task
+    st.session_state["sp_last_ui_task_status"] = str(task.get("status") or "")
+    try:
+        request_path = _task_dir(task_id) / "request.json"
+        if not request_path.is_file():
+            _persist_task_request(task, mode, lang_code)
+        _launch_executor(task_id)
+    except Exception as exc:
+        st.session_state["sp_pending_launch"] = None
+        st.session_state["sp_pending_cancel_task_id"] = task_id
+        st.session_state["sp_terminal_notice"] = {"kind": "error", "message": str(exc)[:800]}
+        st.rerun()
+    st.session_state["sp_pending_launch"] = None
+    st.session_state["sp_payment_required"] = False
+    st.rerun()
 
-    last_error = None
-    for attempt in range(3):
-        try:
-            response = finish_generation_task(
-                task_id,
-                status=status,
-                safe_error_code=safe_error_code[:80],
-                safe_error_detail=safe_error_detail[:800],
-                current_node=current_node[:240],
-            )
-            updated_task = dict(response.get("task") or task)
-            if status == "resumable":
-                st.session_state["sp_generation_task"] = updated_task
-            else:
-                st.session_state["sp_generation_task"] = None
-                st.session_state["sp_generation_idempotency_key"] = ""
-                st.session_state["sp_last_task_heartbeat"] = 0.0
-            return response
-        except Exception as exc:
-            last_error = exc
-            if attempt < 2:
-                time.sleep(0.6 * (attempt + 1))
-
-    raise RuntimeError(f"task_finish_failed: {last_error}")
-
-
-def _quota_start_error_message(language: str, error: Exception) -> str:
-    raw = str(error or "")
-    if "payment_required" not in raw:
-        return raw
-    messages = {
-        "zh": "今日免费额度和长期额度均已用完。支付功能将在下一阶段接入。",
-        "en": "Your daily free quota and lifetime quota are exhausted. Payment will be connected in the next stage.",
-        "ru": "Дневная бесплатная и долгосрочная квоты исчерпаны. Оплата будет подключена на следующем этапе.",
-        "kk": "Күнделікті тегін және ұзақ мерзімді квота таусылды. Төлем келесі кезеңде қосылады.",
-    }
-    return messages.get(language, messages["zh"])
 
 inject_wechat_browser_notice()
 
 DEFAULT_TOPIC = "Generative Artificial Intelligence in Ubiquitous Learning: Evaluating a Chatbot-based Recommendation Engine for Personalized and Context-aware Education"
 DEFAULT_KEYWORDS = "Ubiquitous computing; Context awareness; Deep learning; Personalized educational; Educational technologies; HCI; Recommendation system; LLMs; Generative AI; ChatGPT; Prompting engineering; Few-shot prompting"
 
-_fragment = getattr(st, "fragment", None) or getattr(st, "experimental_fragment", None)
-if _fragment is None:
-    def _fragment(function):
-        return function
-
-
 def _current_language_code() -> str:
     choice = st.session_state.get("interface_language_choice")
     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")),
-)
+if not bool(st.session_state.get("sp_task_recovery_initialized")):
+    recover_generation_task()
+    st.session_state["sp_task_recovery_initialized"] = True
+_user_workspace = os.path.join(WORKSPACE_DIR, "users", _safe_key(authenticated_user.get("id")))
 os.makedirs(_user_workspace, exist_ok=True)
-CHECKPOINT_FILE = os.path.join(_user_workspace, "snapshot.json")
 
 
-@_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",
-                        )
-                    except Exception as exc:
-                        st.error(str(exc))
-                        return
-                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()
-
-progress_bar = st.empty()
-monitor_ui = st.empty()
-
-def render_monitor(agent, phase, current_sec="", details=""):
-    st.session_state["sp_current_node"] = str(current_sec or phase or "")[:240]
-    monitor_ui.markdown(f"""
+
+_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)
+
+
+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))
-        st.stop()
-
-    st.session_state.is_running = True
-    agent = SSciPipelineAgent(global_ds_key, global_s2_key)
-    SAFE_CHUNK_WORDS = agent.chunk_words_cyrillic if target_lang in ["Russian (Русский)", "Kazakh (Қазақша)"] else agent.chunk_words_default
-    qual_engine = get_qualitative_engine(agent)
-    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"]
-            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
+    previous_status = str(st.session_state.get("sp_last_ui_task_status") or str(task.get("status") or ""))
+    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__)
+    st.session_state["sp_last_ui_task_status"] = status
+    _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 = ""
-            
-            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 = 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 = qual_engine.resolve_qualitative_data(topic, keywords, raw_interviews)
-                st.info(tr(lang_code, "qualitative_ready"))
-
-            state = {
-                "topic": topic, "keywords": keywords, "target_lang": target_lang, "words": words,
-                "user_data_md": user_data_md, "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)))
-                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 ""
-            # ==========================================================
-            
-            pb.progress(15, text=tr(lang_code, "outline_progress"))
-            update_heartbeat()
-            plan = agent.generate_outline(topic, keywords, mode=st.session_state.run_mode)
-            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 = clean_markdown_headings(re.sub(r'(?i)^(摘要|abstract|аннотация|түйіндеме)[:\s]*', '', agent.call_deepseek(_translation_system_prompt(target_lang), abstract_anchor, max_tokens=3000)).strip())
-
-            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 = []
-            for sec in structure:
-                sec_name = sec.get("section", "Section")
-                sec_name = re.sub(r'^[\d\.\s一二三四五六七八九十、]+', '', sec_name).strip()
-                
-                is_method_or_result = sec.get("is_method_or_result", False)
-                target_sec_words = int(words * sec.get("weight", 0.1))
-                
-                if target_sec_words > SAFE_CHUNK_WORDS:
-                    num_chunks = max(2, math.ceil(target_sec_words / SAFE_CHUNK_WORDS))
-                    for i in range(num_chunks): 
-                        adaptive_structure.append({"section": sec_name, "display_name": f"{sec_name} (Part {i+1}/{num_chunks})", "target_words": int(target_sec_words/num_chunks), "is_first": i==0, "is_method_or_result": is_method_or_result})
-                else: 
-                    adaptive_structure.append({"section": sec_name, "display_name": sec_name, "target_words": target_sec_words, "is_first": True, "is_method_or_result": is_method_or_result})
-            
-            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.success(tr(lang_code, "restored"))
-            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 = len(adaptive_structure)
-        all_cite_keys = list(agent.citation_map.keys())
-        valid_chunks = [s for s in adaptive_structure if not any(kw in s["section"].lower() for kw in ["abstract", "keyword", "摘要", "关键字", "appendix", "附录"])]
-        allocation_map = {}
-        current_cite_idx = 0
-        
-        for idx, vc in enumerate(valid_chunks):
-            if idx == len(valid_chunks) - 1: 
-                allocation_map[vc["display_name"]] = all_cite_keys[current_cite_idx:]
-            else:
-                num_cites = max(1, math.ceil(len(all_cite_keys) / len(valid_chunks)))
-                allocation_map[vc["display_name"]] = all_cite_keys[current_cite_idx : current_cite_idx + num_cites]
-                current_cite_idx += num_cites
-
-        translation_executor = ThreadPoolExecutor(max_workers=3)
-        previous_task = None
-
-        for i, block in enumerate(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: 
-                remaining_sections -= 1
-                continue 
-                
-            if previous_task:
-                with st.spinner(tr(lang_code, "sync_previous", name=previous_task["name"])):
-                    try:
-                        while not previous_task["future"].done():
-                            update_heartbeat()
-                            time.sleep(1)
-                        trans_text = previous_task["future"].result()
-                        if trans_text:
-                            full_draft_translated += trans_text + "\n\n"
-                    except Exception as e:
-                        pass
-                    
-                    full_draft_en += previous_task["en_text"] + "\n\n"
-                    state["completed_sections"].append(previous_task["name"])
-                    state["full_draft_translated"] = full_draft_translated
-                    state["full_draft_en"] = full_draft_en
-                    state["context_tail"] = previous_task["context_tail"]
-                    state["agent_state"] = agent.export_state()
-                    
-                    os.makedirs(WORKSPACE_DIR, exist_ok=True)
-                    with open(CHECKPOINT_FILE, "w", encoding="utf-8") as f: 
-                        json.dump(state, f, ensure_ascii=False)
-                previous_task = None
-
-            remaining_sections -= 1
-            adjustment = -accumulated_deviation / remaining_sections if remaining_sections > 0 else 0
-            adjusted_target = min(int(SAFE_CHUNK_WORDS * 1.5), max(50, int(sec['target_words'] + adjustment)))
-            
-            is_quarantined = any(kw in sec_title.lower() for kw in ["abstract", "keyword", "摘要", "关键字", "关键词", "conclusion", "结论", "заключение", "қорытынды", "references", "appendix", "附录"])
-            
-            if is_quarantined: 
-                quarantine_rule = "【CRITICAL】: NO IN-TEXT CITATIONS ALLOWED IN THIS SECTION."
-            else:
-                assigned_tags_str = ", ".join(allocation_map.get(display_name, [])) or "any relevant citations"
-                quarantine_rule = f"【CITATION GUIDANCE】: You are strongly encouraged to integrate citations: {assigned_tags_str}. <AVAILABLE_CITATIONS>{refs_text}</AVAILABLE_CITATIONS>"
-            
-            heading_instruction = f"Write the content for the section '{sec_title}'. DO NOT output the main heading '# {sec_title}' itself."
-            memory_block = f"\n【GLOBAL PAPER MEMORY】\n{agent._build_memory_context(sec_title, 2500)}\n" if agent._build_memory_context(sec_title, 2500) else ""
-
-            if hasattr(st.session_state, "rag_engine") and st.session_state.rag_engine is not None:
-                context_snippets = st.session_state.rag_engine.get_relevant_context(sec_title)
-                if context_snippets: 
-                    memory_block += f"\n【TARGETED RAG RESULTS (Includes JIT)】\n{context_snippets}\n"
-
-            lang_hint = "ru" if "Русский" in target_lang else ("kk" if "Қазақша" in target_lang else "en")
-            num_paragraphs = agent._calc_paragraph_count(adjusted_target, lang_hint)
-            
-            if "introduction" in sec_title.lower() or "conclusion" in sec_title.lower():
-                adaptive_heading_rule = "[HEADING MANDATE] DO NOT use any subheadings (no ## or ###) in this section. Write in continuous, well-structured paragraphs."
-            else:
-                adaptive_heading_rule = f"[HEADING MANDATE] Use Level 2 (##) and Level 3 (###) subheadings to structure the text." if adjusted_target > 800 else (f"[HEADING MANDATE] Use Level 2 (##) subheadings." if adjusted_target > 400 else "")
-            
-            adaptive_heading_rule += " STRICT GLOBAL RULE: NEVER use Level 4 (####) headings anywhere in the paper. DO NOT prepend numbers (e.g., 1., 2.1, 一、) to your subheadings."
-
-            table_rule = "[CRITICAL DATA FORMATTING] You MUST present ALL statistical data using STANDARD MARKDOWN TABLES. IMPORTANT: Table titles/captions MUST be normal plain text (e.g., 'Table 1: Data Summary'), DO NOT use heading syntax (like ## or ###) for table titles." if sec["is_method_or_result"] and st.session_state.run_mode == "amj" else ""
-
-            length_directive = f"[LANGUAGE MANDATE] Output strictly in ENGLISH.\n[LENGTH MANDATE - STRICT] Write between {int(adjusted_target * 0.9)} and {int(adjusted_target * 1.1)} words (approx {num_paragraphs} paragraphs).\n{adaptive_heading_rule}\n{table_rule}\n[FINAL CHECK] Do not output a word count statement."
-
-            extra_directives = []
-            lower_sec = sec_title.lower()
-            golden_meta = agent.global_memory.get("core_golden_metadata", [])
-
-            if golden_meta:
-                if any(kw in lower_sec for kw in ["introduction", "background"]): 
-                    extra_directives.append("【TARGETED DIRECTIVE】 Locate research gaps from [TARGETED INJECTION].")
-                elif any(kw in lower_sec for kw in ["literature", "theor", "hypothesis", "review", "framing"]): 
-                    extra_directives.append("【TARGETED DIRECTIVE】 Base conceptual framing on theories from [TARGETED INJECTION].")
-                elif any(kw in lower_sec for kw in ["method", "measure", "data", "sample", "setting"]): 
-                    extra_directives.append("【TARGETED DIRECTIVE】 Adapt analytical design from [TARGETED INJECTION].")
-                elif any(kw in lower_sec for kw in ["result", "analysis", "finding"]): 
-                    extra_directives.append("【ZERO HALLUCINATION】 Translate <USER_ACTUAL_DATA> strictly.")
-                elif any(kw in lower_sec for kw in ["discussion", "conclusion", "implication"]): 
-                    extra_directives.append("【TARGETED DIRECTIVE】 Compare findings with [TARGETED INJECTION].")
-                else: 
-                    extra_directives.append(APP_CONFIG.get("deep_refinement_directive", ""))
-            else:
-                if sec["is_method_or_result"]: 
-                    extra_directives.append(APP_CONFIG.get("methodological_defense_directive", ""))
-                    extra_directives.append(APP_CONFIG.get("empirical_data_logic", ""))
-                if any(kw in lower_sec for kw in ["theory", "theoretical", "hypothesis", "introduction", "framing"]): 
-                    extra_directives.append(APP_CONFIG.get("deep_synthesis_directive", ""))
-                    extra_directives.append(APP_CONFIG.get("hypothesis_development_directive", ""))
-                if "discussion" in lower_sec: 
-                    extra_directives.append(APP_CONFIG.get("discussion_depth_directive", ""))
-                if "conclusion" in lower_sec: 
-                    extra_directives.append(APP_CONFIG.get("conclusion_anchor_directive", ""))
-            
-            theory_injection = "\n\n".join(extra_directives)
-            sys_prompt_m3 = APP_CONFIG.get("sys_academic_writer", "") + "\n" + APP_CONFIG.get("deep_refinement_directive", "")
-            wiz_data_injection = f"\n<USER_ACTUAL_DATA>\n{user_data_md}\n</USER_ACTUAL_DATA>\n" if user_data_md and sec["is_method_or_result"] else ""
-            
-            action_verb = f"Generate the main section '{sec_title}'" if sec["is_first"] else f"Continue generating the subsequent paragraphs for the section '{sec_title}' (DO NOT restate the main section title)"
-            
-            done_list = ", ".join(completed_sections)
-            anti_repeat = f"\n[ANTI-REPETITION MANDATE] You have already generated: [{done_list}]. Do NOT restate core arguments, research gaps, or background introductions belonging to those completed sections." if done_list else ""
-            
-            if "appendix" in sec_title.lower() and user_data_md:
-                base_user = f"Present the following qualitative data structure as a formal Appendix for the paper '{topic}'.\n\nDATA TABLE:\n{user_data_md}\n\nInstruction: Introduce the table with one short paragraph explaining the coding process, then output the exact table. DO NOT write a long analytical section.\n{heading_instruction}\n[FINAL CHECK] Output strictly in ENGLISH."
-            else:
-                base_user = f"{action_verb}.\n{length_directive}\n{quarantine_rule}\n{heading_instruction}\n{theory_injection}\n{anti_repeat}\n{memory_block}\nSynthesis: {synthesis_result}\n{wiz_data_injection}\nPrevious Tail: ...{context_tail}...\nIMPORTANT: Output final text."
-            
-            if agent._count_tokens(sys_prompt_m3) + agent._count_tokens(base_user) > 60000: 
-                base_user = agent._truncate_context(base_user, 60000 - agent._count_tokens(sys_prompt_m3))
-            
-            engine_label = tr(lang_code, "engine_amj") if st.session_state.run_mode == "amj" else tr(lang_code, "engine_asq")
-            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=engine_label))
-            en_ph.empty() 
-            
-            try:
-                with st.status(tr(lang_code, "draft_status", name=display_name), expanded=True) as status:
-                    update_heartbeat()
-                    st.write(tr(lang_code, "api_requested"))
-                    
-                    target_model = "deepseek-v4-flash"
-                    st.write(tr(lang_code, "model_writing", model=target_model))
-                    
-                    en_draft_clean = re.sub(r'<think>.*?</think>', '', en_ph.write_stream(
-                        stream_with_heartbeat(agent.stream_deepseek(sys_prompt_m3, base_user, max_tokens=6000, model_name=target_model, thinking=False))
-                    ), flags=re.DOTALL).strip()
-                    
-                    status.update(label=tr(lang_code, "draft_done", name=display_name), state="complete", expanded=False)
-
-                update_heartbeat()
-                # 1. 移除多余的章节总标题和分块标记
-                en_draft_clean = re.sub(rf'(?im)^#\s*{re.escape(sec_title)}\s*\n+', '', en_draft_clean).strip()
-                en_draft_clean = re.sub(r'(?im)^#+\s*.*\(Part\s*\d+/\d+\).*?\n+', '', en_draft_clean)
-                
-                # 2. 【保留三级标题】保留 ## 和 ###，仅仅去除它们自带的数字编号 (如 "### 3.1.1 样本" 变成 "### 样本")
-                en_draft_clean = re.sub(r'(?m)^(#{2,3})\s+[\d\.\s一二三四五六七八九十、]+\s*(.*)', r'\1 \2', en_draft_clean)
-                
-                # ==========================================================
-                # 3. 【彻底拦截四级标题】精准匹配 4 个或更多的 #，将其强行转为加粗纯文本
-                # ==========================================================
-                en_draft_clean = re.sub(r'(?m)^#{4,}\s+(.*)', r'\n**\1**\n', en_draft_clean)
-                # ==========================================================
-                
-                if "introduction" in sec_title.lower():
-                    en_draft_clean = re.sub(r'(?m)^###?\s+(.*)', r'**\1**', en_draft_clean)
-                    
-                en_draft_clean = re.sub(r'(?m)^###?\s+(\**Table\s+\d+.*)', r'\1', en_draft_clean, flags=re.IGNORECASE)
-
-                if sec["is_first"]: 
-                    en_draft_clean = f"# {sec_title}\n\n{en_draft_clean}"
-                
-                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_draft_clean = en_draft_clean.replace(tag, f'"{quote}"')
-                        
-                en_resolved = agent.resolve_citations(en_draft_clean, quarantine=is_quarantined)
-                en_ph.markdown(en_resolved)
-                
-                agent._extract_memory_from_text(en_draft_clean, sec_title)
-                words_tail = en_draft_clean.split()
-                context_tail = " ".join(words_tail[-200:]) if len(words_tail) > 200 else en_draft_clean
-                
-                actual_words = len(en_draft_clean.split())
-                accumulated_deviation += (actual_words - adjusted_target)
-                st.caption(tr(lang_code, "word_count", actual=actual_words, target=adjusted_target, deviation=accumulated_deviation))
-                
-                current_step += 1
-                pb.progress(min(15 + int((current_step / total_steps) * 80), 95), text=tr(lang_code, "section_complete_progress", name=display_name))
-                
-                trans_ph.info(tr(lang_code, "translation_running"))
-                
-                ctx = get_script_run_ctx()
-                def thread_wrapper(agent_obj, text, ph, lang, q_flag):
-                    add_script_run_ctx(threading.current_thread(), ctx)
-                    return async_translate_task(agent_obj, text, ph, lang, q_flag, lang_code)
-                
-                future = translation_executor.submit(thread_wrapper, agent, en_draft_clean, trans_ph, target_lang, is_quarantined)
-                
-                previous_task = {
-                    "name": display_name, 
-                    "future": future, 
-                    "en_text": en_resolved,
-                    "context_tail": context_tail
-                }
-
-            except Exception as e:
-                en_ph.error(tr(lang_code, "node_error"))
-                st.error(tr(lang_code, "node_error_detail", error=e))
-                generation_aborted = True
-                generation_failure_code = "E_GENERATION_NODE_FAILED"
-                generation_failure_detail = str(e)[:800]
-                break
-
-        if generation_aborted:
-            _finish_managed_generation_task(
-                "resumable",
-                safe_error_code=generation_failure_code or "E_GENERATION_INTERRUPTED",
-                safe_error_detail=generation_failure_detail or "Generation stopped with a recoverable checkpoint.",
-                current_node=str(st.session_state.get("sp_current_node") or "generation"),
-            )
-
-        if not generation_aborted and previous_task:
-            with st.spinner(tr(lang_code, "sync_last", name=previous_task["name"])):
-                try:
-                    while not previous_task["future"].done():
-                        update_heartbeat()
-                        time.sleep(1)
-                    trans_text = previous_task["future"].result()
-                    if trans_text:
-                        full_draft_translated += trans_text + "\n\n"
-                except Exception as e:
-                    pass
-                
-                full_draft_en += previous_task["en_text"] + "\n\n"
-                state["completed_sections"].append(previous_task["name"])
-                state["full_draft_translated"] = full_draft_translated
-                state["full_draft_en"] = full_draft_en
-                state["context_tail"] = previous_task["context_tail"]
-                state["agent_state"] = agent.export_state()
-                
-                with open(CHECKPOINT_FILE, "w", encoding="utf-8") as f: 
-                    json.dump(state, f, ensure_ascii=False)
-
-        if not generation_aborted and full_draft_en.strip():
-            pb.progress(96, text=tr(lang_code, "pack_docx"))
-            update_heartbeat()
-            full_draft_en = clean_trailing_references(full_draft_en, ["Reference", "References", "Reference List"])
-            full_draft_translated = clean_trailing_references(full_draft_translated, ["参考文献", "Reference", "References", "Reference List"])
-            
-            total_words_en = len(full_draft_en.split())
-            target_words = int(words)
-            lower_limit = int(target_words * 0.95)
-            upper_limit = int(target_words * 1.05)
-            st.info(tr(lang_code, "total_words", actual=total_words_en, lower=lower_limit, upper=upper_limit))
-            
-            if total_words_en > upper_limit:
-                conclusion_match = re.search(r'(#+\s*Conclusion.*?)(?=\n#+ |\Z)', full_draft_en, re.DOTALL | re.IGNORECASE)
-                if conclusion_match:
-                    compressed_en = agent.call_deepseek("Condense the following conclusion by 15-20% preserving citations.", conclusion_match.group(1), max_tokens=1500, model_name="deepseek-chat")
-                    if compressed_en:
-                        compressed_en = agent.resolve_citations(compressed_en, quarantine=True)
-                        full_draft_en = full_draft_en.replace(conclusion_match.group(1), compressed_en.strip())
-                        compressed_trans = agent._translate_text(compressed_en, target_lang, max_tokens=2000)
-                        if compressed_trans:
-                            compressed_trans = agent.resolve_citations(compressed_trans, quarantine=True)
-                            trans_match = re.search(r'(#+\s*(?:结论|结论和启示|讨论与结论|Заключение|Қорытынды|Conclusion).*?)(?=\n#+ |\Z)', full_draft_translated, re.DOTALL | re.IGNORECASE)
-                            if trans_match: 
-                                full_draft_translated = full_draft_translated.replace(trans_match.group(1), compressed_trans.strip())
-                            else: 
-                                full_draft_translated += "\n\n" + compressed_trans.strip()
-            elif total_words_en < lower_limit:
-                expanded_en = agent.call_deepseek("Continue conclusion by adding 150 words (future research).", full_draft_en[-2000:], max_tokens=800, model_name="deepseek-chat")
-                if expanded_en:
-                    expanded_en = agent.resolve_citations(expanded_en, quarantine=True)
-                    full_draft_en += "\n\n" + expanded_en.strip()
-                    expanded_trans = agent._translate_text(expanded_en, target_lang, max_tokens=1000)
-                    if expanded_trans: 
-                        expanded_trans = agent.resolve_citations(expanded_trans, quarantine=True)
-                        full_draft_translated += "\n\n" + expanded_trans.strip()
-            
-            if len(agent.used_citations) < 50: 
-                for tag in [t for t in agent.citation_map if t not in agent.used_citations][:50 - len(agent.used_citations)]: 
-                    agent.used_citations.add(tag)
-            
-            apa_refs = "\n\n".join([f"- {ref}" for ref in sorted(list(set([agent.citation_map[t]['ref'] for t in sorted(list(agent.used_citations)) if t in agent.citation_map])))])
-            full_draft_en += "\n\n# References\n" + apa_refs
-            full_draft_translated += f"\n\n# {get_reference_heading(lang_code)}\n" + apa_refs
-
-            bj_time = datetime.now(timezone(timedelta(hours=8))).strftime("%Y%m%d_%H%M")
-            zip_buffer = io.BytesIO()
-            with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
-                zf.writestr(f"Research_EN_{bj_time}.docx", text_to_docx(full_draft_en).getvalue())
-                zf.writestr(f"Research_{output_suffix}_{bj_time}.docx", text_to_docx(full_draft_translated).getvalue())
-            zip_buffer.seek(0)
-            generated_zip_bytes = zip_buffer.getvalue()
-            generated_zip_filename = f"Draft_{bj_time}.zip"
-
-            # Quota is committed only after both DOCX files and the ZIP are complete.
-            completion_finalization_attempted = True
-            _finish_managed_generation_task(
-                "completed",
-                current_node="package_complete",
-            )
-
-            st.session_state.zip_buffer = generated_zip_bytes
-            st.session_state.zip_filename = generated_zip_filename
-
-            try:
-                disk_save_path = os.path.join(OUTPUT_DIR, generated_zip_filename)
-                with open(disk_save_path, "wb") as f_out:
-                    f_out.write(generated_zip_bytes)
-
-                saved_files = sorted([os.path.join(OUTPUT_DIR, f) for f in os.listdir(OUTPUT_DIR) if f.endswith('.zip')], key=os.path.getmtime)
-                while len(saved_files) > 5:
-                    os.remove(saved_files.pop(0))
-            except Exception:
-                pass
-
-            try:
-                os.remove(CHECKPOINT_FILE)
-            except OSError:
-                pass
-
-            pb.progress(100, text=tr(lang_code, "complete"))
-            render_monitor(agent, tr(lang_code, "complete"), "✅", tr(lang_code, "document_generated"))
-            
-    except Exception as e:
-        if not completion_finalization_attempted:
-            checkpoint_available = _load_recoverable_checkpoint() is not None
-            try:
-                _finish_managed_generation_task(
-                    "resumable" if checkpoint_available else "failed",
-                    safe_error_code="E_GENERATION_INTERRUPTED" if checkpoint_available else "E_GENERATION_FAILED",
-                    safe_error_detail=str(e)[:800],
-                    current_node=str(st.session_state.get("sp_current_node") or "generation"),
-                )
-            except Exception:
-                pass
-        st.error(tr(lang_code, "startup_error", error=e))
-    finally:
-        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)
+
+    # Only boundary transitions need one full app rerun so static controls pick
+    # up resume/terminal state. Running progress never reruns the whole page.
+    if previous_status and status != previous_status and status in {"resumable", "completed", "failed", "cancelled"}:
+        st.rerun()
+
+
+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

@@ -4,22 +4,38 @@ import os
 def load_config():
     base_config = {
         # --- 基础运行参数 ---
-        "lit_target_papers": int(os.environ.get("LIT_TARGET_PAPERS", "250")),
+        "lit_target_papers": int(os.environ.get("LIT_TARGET_PAPERS", "80")),
         "lit_per_page": int(os.environ.get("LIT_PER_PAGE", "100")),
         "lit_max_workers": int(os.environ.get("LIT_MAX_WORKERS", "5")),
-        "lit_max_page_batches": int(os.environ.get("LIT_MAX_PAGE_BATCHES", "12")),
         "chunk_words_default": int(os.environ.get("CHUNK_WORDS_DEFAULT", "1500")),
-        "chunk_words_cyrillic": int(os.environ.get("CHUNK_WORDS_CYRILLIC", "1000")),
+        "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"))),
+        "translation_min_coverage_ratio": float(os.environ.get("TRANSLATION_MIN_COVERAGE_RATIO", "0.25")),
+        "translation_min_block_ratio_zh": float(os.environ.get("TRANSLATION_MIN_BLOCK_RATIO_ZH", "0.28")),
+        "translation_min_block_ratio_cyrillic": float(os.environ.get("TRANSLATION_MIN_BLOCK_RATIO_CYRILLIC", "0.65")),
+        "translation_max_block_ratio": float(os.environ.get("TRANSLATION_MAX_BLOCK_RATIO", "2.20")),
+        "openalex_email": os.environ.get("OPENALEX_EMAIL", "").strip(),
         
         # --- RAG 与解析配置 ---
         "rag_chunk_size": int(os.environ.get("RAG_CHUNK_SIZE", "3000")),
         "rag_top_k": int(os.environ.get("RAG_TOP_K", "3")),
         "max_extract_chars": int(os.environ.get("MAX_EXTRACT_CHARS", "150000")),
+        # A DeepSeek request must fail before the 90-second Worker generation
+        # lease can expire without real execution progress. Retries renew the
+        # lease through the execution heartbeat hook.
         "api_timeouts": {
-            "standard": (15, 180),
-            "stream": (15, 300),
+            "standard": (15, 70),
+            "stream": (15, 70),
             "search": 15
         },
+        "deepseek_stream_idle_seconds": max(
+            20,
+            min(65, int(os.environ.get("DEEPSEEK_STREAM_IDLE_SECONDS", "60"))),
+        ),
         
         # --- API 节点配置 ---
         "deepseek_api_url": os.environ.get(
@@ -82,8 +98,9 @@ def load_config():
             "DEEPSEEK_SHOW_REASONING",
             "false",
         ).strip().lower() in {"1", "true", "yes", "on"},
-        # Citation-semantic audit is fail-closed. It reviews only sentences
-        # containing source tags and never silently deletes extra sources.
+        # Citation-semantic audit repairs only affected sentences. Technical
+        # failures fall back to deterministic single-source recovery and are
+        # recorded in the audit log instead of terminating the whole paper.
         "citation_semantic_audit": {
             "enabled": os.environ.get(
                 "CITATION_SEMANTIC_AUDIT_ENABLED",
@@ -99,7 +116,7 @@ def load_config():
             ),
             "fail_closed": os.environ.get(
                 "CITATION_SEMANTIC_FAIL_CLOSED",
-                "true",
+                "false",
             ).strip().lower() in {"1", "true", "yes", "on"},
             "technical_retries": max(
                 1,
@@ -110,18 +127,42 @@ 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")),
         ),
+        "executor_bootstrap_timeout_seconds": max(
+            5,
+            int(os.environ.get("SMARTPAPER_EXECUTOR_BOOTSTRAP_TIMEOUT_SECONDS", "20")),
+        ),
+        "executor_bootstrap_poll_ms": max(
+            50,
+            int(os.environ.get("SMARTPAPER_EXECUTOR_BOOTSTRAP_POLL_MS", "100")),
+        ),
+        "manuscript_quality": {
+            "abstract_max_words": int(os.environ.get("ABSTRACT_MAX_WORDS", "200")),
+            "abstract_max_numeric_mentions": int(os.environ.get("ABSTRACT_MAX_NUMERIC_MENTIONS", "6")),
+            "forbidden_phrases": [
+                "delve into", "intricate", "paramount", "overarching",
+                "pivotal", "foster", "seamlessly", "multifaceted",
+                "beacon", "landscape", "underscore", "robust",
+                "firstly", "secondly", "moreover", "furthermore",
+                "in conclusion", "to summarize", "as an ai",
+                "ai-generated", "this response",
+            ],
+        },
         "openalex_api_url": os.environ.get("OPENALEX_API_URL", "https://api.openalex.org/works"),
         "s2_api_url": os.environ.get("S2_API_URL", "https://api.semanticscholar.org/graph/v1"),
         
         # --- 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": "🎉 Тапсырма аяқталды!"}
@@ -331,7 +372,7 @@ Literature Metadata:
 [ABSOLUTE ANTI-AI-TONE MANDATES]:
 1. CRITICAL: NEVER use these cliché AI words: delve into, intricate, paramount, overarching, pivotal, foster, seamlessly, multifaceted, beacon, landscape, underscore, robust.
 2. CRITICAL: NEVER use robotic transition phrases (e.g., Firstly, Moreover, Furthermore, In conclusion, To summarize). Use conceptual bridges instead.
-3. SYNTAX BURSTINESS: You MUST vary sentence length dramatically. Mix short, punchy, definitive claims with longer, nuanced analytical sentences.
+3. NATURAL SYNTAX: Vary sentence structure only where the argument requires it. Avoid artificial short-long sentence alternation and repetitive templates.
 4. ACADEMIC HEDGING: Use strategic hedging (e.g., 'This suggests a tendency toward X, bounded by Y').
 5. NO SWEEPING GENERALIZATIONS: Every sentence must carry specific theoretical weight or empirical fact.
 6. FOCUS ON CAUSALITY: You MUST articulate the exact theoretical mechanism driving the relationship.

core/checkpoint_store.py

@@ -0,0 +1,71 @@
+# -*- coding: utf-8 -*-
+"""Lightweight task-scoped checkpoint persistence for SmartPaper executors.
+
+This module intentionally depends only on the Python standard library. It can
+be used immediately after a Worker claim without importing DeepSeek, RAG,
+statistics, document, or Streamlit modules.
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+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)
+
+
+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)
+        _atomic_json(self.directory / "latest.json", payload)
+        return self.sequence, stage_name

core/desktop.ini

@@ -0,0 +1,2 @@
+[.ShellClassInfo]
+IconResource=C:\Program Files\Google\Drive File Stream\129.0.1.0\GoogleDriveFS.exe,27

core/generation_runner.py

@@ -0,0 +1,536 @@
+# -*- 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.checkpoint_store import TaskCheckpointStore
+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 _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 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
@@ -13,7 +12,8 @@ import sys
 import logging
 import concurrent.futures
 from collections import Counter
-from tenacity import retry, stop_after_attempt, wait_random_exponential, retry_if_exception_type
+from datetime import datetime, timezone
+from email.utils import parsedate_to_datetime
 
 # 基础日志配置：自带时间戳和日志级别
 logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[logging.StreamHandler(sys.stdout)])
@@ -62,15 +62,12 @@ 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
         # Backward-compatible counters now mirror unique additions, while the
@@ -82,27 +79,30 @@ class SSciPipelineAgent:
                 "request_count": 0,
                 "success_count": 0,
                 "raw_result_count": 0,
+                "eligible_result_count": 0,
+                "merged_count": 0,
                 "unique_added_count": 0,
                 "duplicate_count": 0,
                 "error_count": 0,
                 "last_status_code": None,
                 "last_safe_error": "",
+                "status": "not_started",
             }
             for source in ("s2", "oa")
         }
         self._cancel_checker = None
+        self._execution_progress_hook = None
         self.s2_fused = False
         self.oa_fused = False
         self.citation_map = {}
         self.used_citations = set()
+        self.citation_audit_log = []
         self.lock = threading.RLock()
         
         self.lit_target_papers = APP_CONFIG["lit_target_papers"]
         self.lit_per_page = APP_CONFIG["lit_per_page"]
         self.lit_max_workers = APP_CONFIG["lit_max_workers"]
-        self.lit_max_page_batches = APP_CONFIG["lit_max_page_batches"]
         self.chunk_words_default = APP_CONFIG["chunk_words_default"]
-        self.chunk_words_cyrillic = APP_CONFIG["chunk_words_cyrillic"]
         
         self.deepseek_api_url = APP_CONFIG["deepseek_api_url"]
         configured_models = APP_CONFIG.get("deepseek_models") or {}
@@ -143,6 +143,10 @@ class SSciPipelineAgent:
         self.openalex_api_url = APP_CONFIG["openalex_api_url"]
         self.s2_api_url = APP_CONFIG["s2_api_url"]
         self.timeouts = APP_CONFIG["api_timeouts"]
+        self.deepseek_stream_idle_seconds = max(
+            20,
+            int(APP_CONFIG.get("deepseek_stream_idle_seconds", 60)),
+        )
         
         self.global_memory = {
             "hypotheses": "", "variables": "", "model_summary": "",
@@ -160,6 +164,23 @@ class SSciPipelineAgent:
     def set_cancel_checker(self, checker):
         self._cancel_checker = checker
 
+    def set_execution_progress_hook(self, hook):
+        """Register a best-effort callback tied to real backend execution."""
+        self._execution_progress_hook = hook
+
+    def _touch_execution_progress(self, stage=""):
+        hook = self._execution_progress_hook
+        if hook is None:
+            return
+        try:
+            hook()
+        except Exception as exc:
+            logging.warning(
+                "Execution heartbeat hook failed | stage=%s | error=%s",
+                str(stage or "unknown")[:80],
+                type(exc).__name__,
+            )
+
     def _check_cancelled(self):
         if self._cancel_checker is not None:
             result = self._cancel_checker()
@@ -228,6 +249,7 @@ class SSciPipelineAgent:
                 "source_metrics": self.get_source_metrics(),
                 "citation_map": self.citation_map, 
                 "used_citations": list(self.used_citations),
+                "citation_audit_log": list(self.citation_audit_log),
                 "global_memory": self.global_memory
             }
 
@@ -244,47 +266,96 @@ class SSciPipelineAgent:
                     self.source_metrics[source].update(loaded_metrics[source])
             self.citation_map = state_dict.get("citation_map", {})
             self.used_citations = set(state_dict.get("used_citations", []))
+            self.citation_audit_log = list(state_dict.get("citation_audit_log", []))
             self.global_memory = state_dict.get("global_memory", self.global_memory)
         logging.info(f"🔄 状态恢复成功: DS API={self.ds_calls}, 现有文献={len(self.citation_map)}")
 
-    def _calc_paragraph_count(self, target_words, lang_hint="en"):
-        words_per_para = 100 if lang_hint in ["ru", "kk"] else 130
-        return max(2, int(target_words / words_per_para))
-
-    def _translate_text(self, text, target_lang, max_tokens=2000):
-        if not text.strip(): 
-            return ""
-        sys_prompt = f"Translate the following text into {target_lang}. Preserve formatting and [CITE-XX] tags exactly. Output ONLY the translation."
-        return self.call_deepseek(
-            sys_prompt,
-            text,
-            max_tokens=max_tokens,
-            temp=0.2,
-            model_role="translation",
-            thinking=False,
-            allow_fallback=False,
-        )
-
-    @retry(stop=stop_after_attempt(4), wait=wait_random_exponential(multiplier=1.5, max=15), retry=retry_if_exception_type((requests.exceptions.Timeout, requests.exceptions.ConnectionError, requests.exceptions.HTTPError)))
     def _requests_get_with_retry(self, url, headers=None, params=None, timeout=None):
-        self._check_cancelled()
-        if timeout is None: 
+        """GET with bounded, status-aware retry semantics.
+
+        Authentication and query errors are returned immediately. Rate limits
+        honour ``Retry-After`` when present. Only timeouts, connection failures,
+        429, and 5xx responses are retried.
+        """
+        if timeout is None:
             timeout = self.timeouts["search"]
-        resp = requests.get(url, headers=headers, params=params, timeout=timeout)
-        self._check_cancelled()
-        
-        if resp.status_code == 429:
-            logging.warning(f"⚠️ [API 限流] 触发 429 (Too Many Requests)。启动指数退避。URL: {url}")
-            time.sleep(random.uniform(2.0, 5.0))
-            resp.raise_for_status()
-            
-        if resp.status_code == 400 and "Requested data for this limit" in resp.text: 
-            return resp
-            
-        if resp.status_code != 200:
-            logging.error(f"❌ [API 错误] 状态码: {resp.status_code} | URL: {url}")
-            resp.raise_for_status()
-        return resp
+        attempts = max(1, int(APP_CONFIG.get("search_max_retries", 4)))
+        last_error = None
+        for attempt in range(attempts):
+            self._check_cancelled()
+            try:
+                response = requests.get(
+                    url,
+                    headers=headers,
+                    params=params,
+                    timeout=timeout,
+                )
+            except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as error:
+                last_error = error
+                if attempt + 1 >= attempts:
+                    raise
+                delay = min(12.0, 1.5 * (2 ** attempt)) + random.uniform(0.0, 0.5)
+                logging.warning(
+                    "Search transport retry %s/%s after %s: %s",
+                    attempt + 1,
+                    attempts,
+                    type(error).__name__,
+                    url,
+                )
+                time.sleep(delay)
+                continue
+
+            self._check_cancelled()
+            status = int(response.status_code)
+            if status in (400, 401, 403):
+                return response
+            if status == 429:
+                if attempt + 1 >= attempts:
+                    return response
+                retry_after = str(response.headers.get("Retry-After") or "").strip()
+                delay = None
+                if retry_after:
+                    try:
+                        delay = float(retry_after)
+                    except ValueError:
+                        try:
+                            retry_at = parsedate_to_datetime(retry_after)
+                            if retry_at.tzinfo is None:
+                                retry_at = retry_at.replace(tzinfo=timezone.utc)
+                            delay = (retry_at - datetime.now(timezone.utc)).total_seconds()
+                        except (TypeError, ValueError, OverflowError):
+                            delay = None
+                if delay is None:
+                    delay = 2.0 * (2 ** attempt) + random.uniform(0.0, 0.5)
+                delay = max(1.0, min(30.0, delay))
+                logging.warning(
+                    "Search rate-limit retry %s/%s in %.1fs: %s",
+                    attempt + 1,
+                    attempts,
+                    delay,
+                    url,
+                )
+                time.sleep(delay)
+                continue
+            if 500 <= status <= 599:
+                if attempt + 1 >= attempts:
+                    response.raise_for_status()
+                delay = min(12.0, 1.5 * (2 ** attempt)) + random.uniform(0.0, 0.5)
+                logging.warning(
+                    "Search server retry %s/%s after HTTP %s: %s",
+                    attempt + 1,
+                    attempts,
+                    status,
+                    url,
+                )
+                time.sleep(delay)
+                continue
+            response.raise_for_status()
+            return response
+
+        if last_error is not None:
+            raise last_error
+        raise requests.exceptions.RequestException("search_request_failed")
 
     def _is_english_dominant(self, text):
         if not text: 
@@ -654,6 +725,13 @@ class SSciPipelineAgent:
 
             try:
                 self._check_cancelled()
+                self._touch_execution_progress("deepseek_static_request")
+                logging.info(
+                    "DeepSeek static attempt start | attempt=%s/%s | model=%s",
+                    attempt + 1,
+                    self.deepseek_max_retries,
+                    current_model,
+                )
                 response = requests.post(
                     self.deepseek_api_url,
                     headers=self._deepseek_headers(self.ds_key),
@@ -662,6 +740,7 @@ class SSciPipelineAgent:
                 )
 
                 self._check_cancelled()
+                self._touch_execution_progress("deepseek_static_response")
                 if response.status_code != 200:
                     self._raise_for_deepseek_status(response)
 
@@ -715,6 +794,7 @@ class SSciPipelineAgent:
                     self.deepseek_max_retries,
                     exc,
                 )
+                self._touch_execution_progress("deepseek_static_retry")
                 if is_last:
                     break
                 time.sleep(self._retry_delay(attempt))
@@ -790,10 +870,22 @@ class SSciPipelineAgent:
             emitted_content = False
             reasoning_open = False
             reasoning_chunks = 0
+            content_chunks = 0
+            content_chars = 0
+            first_content_at = None
             start_time = time.time()
+            last_model_activity = start_time
 
             try:
                 self._check_cancelled()
+                self._touch_execution_progress("deepseek_stream_request")
+                logging.info(
+                    "DeepSeek stream attempt start | attempt=%s/%s | model=%s | idle_limit=%ss",
+                    attempt + 1,
+                    self.deepseek_max_retries,
+                    current_model,
+                    self.deepseek_stream_idle_seconds,
+                )
                 with requests.post(
                     self.deepseek_api_url,
                     headers=self._deepseek_headers(self.ds_key),
@@ -809,9 +901,22 @@ class SSciPipelineAgent:
                     ):
                         self._check_cancelled()
                         event = self._parse_sse_line(raw_line)
+                        now = time.time()
+                        if now - last_model_activity > self.deepseek_stream_idle_seconds:
+                            raise DeepSeekAPIError(
+                                "DeepSeek stream produced no model progress for "
+                                f"{self.deepseek_stream_idle_seconds} seconds."
+                            )
                         if not event:
                             continue
                         if event.get("done"):
+                            logging.info(
+                                "DeepSeek stream done marker | seconds=%.1f | reasoning_chunks=%s | content_chunks=%s | content_chars=%s",
+                                now - start_time,
+                                reasoning_chunks,
+                                content_chunks,
+                                content_chars,
+                            )
                             break
 
                         choices = event.get("choices") or []
@@ -825,7 +930,9 @@ class SSciPipelineAgent:
                         content = delta.get("content") or ""
 
                         if reasoning_content:
+                            last_model_activity = time.time()
                             reasoning_chunks += 1
+                            self._touch_execution_progress("deepseek_reasoning")
                             if not reasoning_open:
                                 reasoning_open = True
                                 logging.info(
@@ -848,6 +955,16 @@ class SSciPipelineAgent:
                                 )
 
                         if content:
+                            last_model_activity = time.time()
+                            self._touch_execution_progress("deepseek_content")
+                            if first_content_at is None:
+                                first_content_at = last_model_activity
+                                logging.info(
+                                    "DeepSeek first content | seconds=%.1f",
+                                    first_content_at - start_time,
+                                )
+                            content_chunks += 1
+                            content_chars += len(str(content))
                             if (
                                 reasoning_open
                                 and self.deepseek_show_reasoning
@@ -865,9 +982,13 @@ class SSciPipelineAgent:
                         "DeepSeek stream ended without final content."
                     )
 
+                self._touch_execution_progress("deepseek_stream_complete")
                 logging.info(
-                    "DeepSeek stream complete | seconds=%.1f",
+                    "DeepSeek stream complete | seconds=%.1f | reasoning_chunks=%s | content_chunks=%s | content_chars=%s",
                     time.time() - start_time,
+                    reasoning_chunks,
+                    content_chunks,
+                    content_chars,
                 )
                 return
 
@@ -899,6 +1020,7 @@ class SSciPipelineAgent:
                         "content had already been emitted."
                     ) from exc
 
+                self._touch_execution_progress("deepseek_stream_retry")
                 is_last = attempt == self.deepseek_max_retries - 1
                 if is_last:
                     break
@@ -1347,15 +1469,27 @@ You MUST output your response EXACTLY using the following two delimiters. Do not
                 results = future.result()
                 with self.lock:
                     for res in results:
-                        if not any(res["id"] == self.citation_map[k].get("id", "") for k in self.citation_map):
-                            cite_tag = f"[CITE-{len(self.citation_map)+1:02d}]"
-                            self.citation_map[cite_tag] = {
-                                "in_text": res["in_text"],
-                                "ref": res["ref"],
-                                "id": res["id"],
-                                "evidence": res.get("rag_text", ""),
-                            }
-                            jit_rag_texts.append(f"{cite_tag} {res['rag_text']}")
+                        result_keys = self._reference_keys(res)
+                        existing_keys = {
+                            key
+                            for metadata in self.citation_map.values()
+                            for key in self._reference_keys(metadata)
+                        }
+                        if result_keys and any(key in existing_keys for key in result_keys):
+                            continue
+                        cite_tag = f"[CITE-{len(self.citation_map)+1:02d}]"
+                        self.citation_map[cite_tag] = {
+                            "in_text": res["in_text"],
+                            "ref": res["ref"],
+                            "id": result_keys[0] if result_keys else ('title_year', res.get('title'), res.get('year')),
+                            "doi": res.get("doi", ""),
+                            "external_id": res.get("external_id", ""),
+                            "title": res.get("title", ""),
+                            "year": res.get("year", ""),
+                            "evidence": res.get("rag_text", ""),
+                            "sources": [res.get("source", "jit")],
+                        }
+                        jit_rag_texts.append(f"{cite_tag} {res.get('rag_text', '')}")
 
         logging.info(f"✅ [Pre-fetch JIT] 成功抓取并注入了 {len(jit_rag_texts)} 篇底层支撑文献。")
         return "\n".join(jit_rag_texts)
@@ -1387,51 +1521,139 @@ Return ONLY a valid JSON array of strings."""
         clean_chunks = [c.strip() for c in re.findall(r'[a-zA-Z0-9\s\-]{4,}', f"{topic} {keywords}") if self._is_english_dominant(c.strip())]
         return clean_chunks[:3] if clean_chunks else []
 
+    @staticmethod
+    def _normalize_search_query(value):
+        query = re.sub(r'["“”‘’()\[\]{}]', ' ', str(value or ''))
+        query = re.sub(r'(?i)\b(?:AND|OR|NOT)\b', ' ', query)
+        query = query.replace('-', ' ')
+        query = re.sub(r'\s+', ' ', query).strip()
+        terms = re.findall(r"[A-Za-z0-9][A-Za-z0-9']*", query)
+        return ' '.join(terms[:8])
+
+    @staticmethod
+    def _normalize_doi(value):
+        doi = str(value or '').strip().lower()
+        doi = re.sub(r'^https?://(?:dx\.)?doi\.org/', '', doi)
+        return doi.strip()
+
+    @classmethod
+    def _reference_keys(cls, result):
+        keys = []
+        doi = cls._normalize_doi(result.get('doi'))
+        if doi:
+            keys.append(('doi', doi))
+        external_id = str(result.get('external_id') or '').strip().lower()
+        if external_id:
+            keys.append(('external', external_id))
+        title = re.sub(r'[^a-z0-9]+', ' ', str(result.get('title') or '').lower()).strip()
+        year = str(result.get('year') or '').strip()
+        if title:
+            keys.append(('title_year', title, year))
+        return keys
+
+    def _record_source_error(self, source, code, status_code=None, detail=''):
+        self._source_metric_add(source, 'error_count')
+        self._source_metric_set(source, 'status', 'error')
+        self._source_metric_set(source, 'last_status_code', status_code)
+        safe = code if not detail else f"{code}: {self._safe_source_error(detail)}"
+        self._source_metric_set(source, 'last_safe_error', safe[:300])
+
+    def _merge_reference_results(self, source_results, limit):
+        """Merge deterministically while preserving each source's contribution.
+
+        ``merged_count`` means that a source is represented in the final merged
+        evidence set, including overlap confirmed by both services.
+        ``unique_added_count`` records which source first introduced a record.
+        """
+        merged = []
+        key_to_index = {}
+        source_lists = {source: list(source_results.get(source, [])) for source in ('s2', 'oa')}
+        max_length = max((len(items) for items in source_lists.values()), default=0)
+        for position in range(max_length):
+            for source in ('s2', 'oa'):
+                items = source_lists[source]
+                if position >= len(items):
+                    continue
+                result = items[position]
+                keys = self._reference_keys(result)
+                duplicate_index = next((key_to_index[k] for k in keys if k in key_to_index), None)
+                if duplicate_index is not None:
+                    self._source_metric_add(source, 'duplicate_count')
+                    sources = merged[duplicate_index].setdefault('sources', [])
+                    if source not in sources:
+                        sources.append(source)
+                        self._source_metric_add(source, 'merged_count')
+                    continue
+                if len(merged) >= limit:
+                    continue
+                item = dict(result)
+                item['sources'] = [source]
+                merged.append(item)
+                index = len(merged) - 1
+                for key in keys:
+                    key_to_index[key] = index
+                self._source_metric_add(source, 'merged_count')
+                self._source_metric_add(source, 'unique_added_count')
+        # The compact, approved UI displays successful eligible returns, not
+        # source ownership after cross-source deduplication. Detailed overlap and
+        # merge counts remain available in LITERATURE_METRICS.json.
+        self.s2_calls = int(self.source_metrics['s2']['eligible_result_count'])
+        self.oa_calls = int(self.source_metrics['oa']['eligible_result_count'])
+        return merged
+
     def _s2_worker(self, term, offset):
         self._check_cancelled()
-        time.sleep(random.uniform(0.5, 1.5))
-        headers = {"x-api-key": str(self.s2_key).strip()} if self.s2_key else {}
-        results = []
-        self._source_metric_add("s2", "request_count")
-        logging.info("S2 search: term=%r offset=%s", term, offset)
+        query = self._normalize_search_query(term)
+        if not query:
+            self._record_source_error('s2', 'S2_BAD_QUERY')
+            return []
+        headers = {'x-api-key': str(self.s2_key).strip()} if self.s2_key else {}
+        self._source_metric_add('s2', 'request_count')
+        self._source_metric_set('s2', 'status', 'running')
+        logging.info('S2 search: term=%r normalized=%r offset=%s', term, query, offset)
         try:
-            resp = self._requests_get_with_retry(
+            response = self._requests_get_with_retry(
                 f"{self.s2_api_url}/paper/search",
                 headers=headers,
                 params={
-                    "query": term,
-                    "offset": offset,
-                    "limit": 50,
-                    "fields": "title,authors,year,venue,abstract,externalIds",
+                    'query': query,
+                    'offset': offset,
+                    'limit': 50,
+                    'fields': 'paperId,title,authors,year,venue,abstract,externalIds',
                 },
             )
-            self._source_metric_set("s2", "last_status_code", resp.status_code)
-            if resp.status_code == 400:
-                self._source_metric_add("s2", "error_count")
-                self._source_metric_set("s2", "last_safe_error", "HTTP 400")
-                return results
-            self._source_metric_add("s2", "success_count")
-            papers = resp.json().get("data", [])
-            self._source_metric_add("s2", "raw_result_count", len(papers))
+            status = response.status_code
+            self._source_metric_set('s2', 'last_status_code', status)
+            if status in (401, 403):
+                self._record_source_error('s2', 'S2_AUTH_ERROR', status)
+                return []
+            if status == 429:
+                self._record_source_error('s2', 'S2_RATE_LIMIT', status)
+                return []
+            if status == 400:
+                self._record_source_error('s2', 'S2_BAD_QUERY', status)
+                return []
+            response.raise_for_status()
+            self._source_metric_add('s2', 'success_count')
+            papers = response.json().get('data', [])
+            self._source_metric_add('s2', 'raw_result_count', len(papers))
+            results = []
             for paper in papers:
                 self._check_cancelled()
-                title = paper.get("title")
-                year = paper.get("year")
-                if not title or not year:
+                title = str(paper.get('title') or '').strip()
+                year = paper.get('year')
+                if not title or not year or not self._is_english_dominant(title):
                     continue
-                authors = paper.get("authors", [])
+                authors = paper.get('authors') or []
                 last_names, formatted_authors = [], []
                 for author in authors:
-                    name = author.get("name", "")
+                    name = str(author.get('name') or '').strip()
                     if not name:
                         continue
                     parts = name.split()
-                    if len(parts) > 1:
-                        last_names.append(parts[-1])
-                        formatted_authors.append(f"{parts[-1]}, {parts[0][0]}.")
-                    else:
-                        last_names.append(name)
-                        formatted_authors.append(name)
+                    last = parts[-1]
+                    last_names.append(last)
+                    formatted_authors.append(f"{last}, {parts[0][0]}." if len(parts) > 1 else last)
                 if not last_names:
                     continue
                 if len(last_names) == 1:
@@ -1440,81 +1662,96 @@ Return ONLY a valid JSON array of strings."""
                     in_text = f"({last_names[0]} & {last_names[1]}, {year})"
                 else:
                     in_text = f"({last_names[0]} et al., {year})"
-                if len(formatted_authors) > 2:
-                    author_str = ", ".join(formatted_authors[:-1]) + ", & " + formatted_authors[-1]
-                elif len(formatted_authors) == 2:
-                    author_str = " & ".join(formatted_authors)
-                else:
-                    author_str = formatted_authors[0]
-                apa_full = f"{author_str} ({year}). {title}. {paper.get('venue') or 'Journal'}."
-                doi = paper.get("externalIds", {}).get("DOI", "")
+                author_str = (', '.join(formatted_authors[:-1]) + ', & ' + formatted_authors[-1]) if len(formatted_authors) > 2 else (' & '.join(formatted_authors) if len(formatted_authors) == 2 else formatted_authors[0])
+                doi = self._normalize_doi((paper.get('externalIds') or {}).get('DOI'))
+                ref = f"{author_str} ({year}). {title}. {paper.get('venue') or 'Journal'}."
                 if doi:
-                    apa_full += f" https://doi.org/{doi}"
-                abstract_str = " ".join((paper.get("abstract") or "").split()[:60])
+                    ref += f" https://doi.org/{doi}"
+                abstract = ' '.join(str(paper.get('abstract') or '').split()[:180])
                 results.append({
-                    "id": (title.lower().strip(), str(year)),
-                    "in_text": in_text,
-                    "ref": apa_full,
-                    "rag_text": f"Title: {title}. Focus: {abstract_str}...",
-                    "source": "s2",
+                    'title': title,
+                    'year': str(year),
+                    'doi': doi,
+                    'external_id': str(paper.get('paperId') or ''),
+                    'in_text': in_text,
+                    'ref': ref,
+                    'rag_text': f"Title: {title}. Abstract evidence: {abstract}",
+                    'source': 's2',
                 })
+            self._source_metric_add('s2', 'eligible_result_count', len(results))
+            self._source_metric_set('s2', 'status', 'ok' if results else 'no_results')
+            if not results:
+                self._source_metric_set('s2', 'last_safe_error', 'S2_NO_RESULTS')
+            return results
         except PipelineCancelledError:
             raise
+        except requests.Timeout as error:
+            self._record_source_error('s2', 'S2_TIMEOUT', detail=error)
         except Exception as error:
-            self._source_metric_add("s2", "error_count")
-            status = getattr(getattr(error, "response", None), "status_code", None)
-            if status is not None:
-                self._source_metric_set("s2", "last_status_code", status)
-            safe_error = self._safe_source_error(error)
-            self._source_metric_set("s2", "last_safe_error", safe_error)
-            logging.warning("S2 search error: %s", safe_error)
-        return results
+            status = getattr(getattr(error, 'response', None), 'status_code', None)
+            code = 'S2_RATE_LIMIT' if status == 429 else ('S2_AUTH_ERROR' if status in (401, 403) else 'S2_REQUEST_ERROR')
+            self._record_source_error('s2', code, status, error)
+        return []
 
     def _oa_worker(self, term, page, filter_str):
         self._check_cancelled()
-        time.sleep(random.uniform(1.0, 2.5))
-        headers = {"User-Agent": f"mailto:ssci_researcher_{random.randint(10000, 99999)}@gmail.com"}
+        query = self._normalize_search_query(term)
+        if not query:
+            self._record_source_error('oa', 'OPENALEX_BAD_QUERY')
+            return []
+        user_agent = 'SmartPaper/production-v1.1'
+        if self.openalex_email:
+            user_agent += f' (mailto:{self.openalex_email})'
+        headers = {'User-Agent': user_agent}
         if self.openalex_key:
-            headers["Authorization"] = f"Bearer {self.openalex_key}"
-        results = []
-        self._source_metric_add("oa", "request_count")
-        logging.info("OpenAlex search: term=%r page=%s", term, page)
+            headers['Authorization'] = f"Bearer {self.openalex_key}"
+        self._source_metric_add('oa', 'request_count')
+        self._source_metric_set('oa', 'status', 'running')
+        logging.info('OpenAlex search: term=%r normalized=%r page=%s', term, query, page)
         try:
-            # Query and authentication parameters intentionally remain unchanged.
-            params = {
-                "search": term,
-                "per-page": self.lit_per_page,
-                "page": page,
-                "filter": filter_str,
-                "mailto": "ssci_researcher_agent@gmail.com",
-            }
-            resp = self._requests_get_with_retry(
+            response = self._requests_get_with_retry(
                 self.openalex_api_url,
                 headers=headers,
-                params=params,
+                params={
+                    **{
+                        'search': query,
+                        'per-page': self.lit_per_page,
+                        'page': page,
+                        'filter': filter_str,
+                    },
+                    **({'mailto': self.openalex_email} if self.openalex_email else {}),
+                },
             )
-            self._source_metric_set("oa", "last_status_code", resp.status_code)
-            self._source_metric_add("oa", "success_count")
-            raw_items = resp.json().get("results", [])
-            self._source_metric_add("oa", "raw_result_count", len(raw_items))
+            status = response.status_code
+            self._source_metric_set('oa', 'last_status_code', status)
+            if status in (401, 403):
+                self._record_source_error('oa', 'OPENALEX_AUTH_ERROR', status)
+                return []
+            if status == 429:
+                self._record_source_error('oa', 'OPENALEX_RATE_LIMIT', status)
+                return []
+            if status == 400:
+                self._record_source_error('oa', 'OPENALEX_BAD_QUERY', status)
+                return []
+            response.raise_for_status()
+            self._source_metric_add('oa', 'success_count')
+            raw_items = response.json().get('results', [])
+            self._source_metric_add('oa', 'raw_result_count', len(raw_items))
+            results = []
             for item in raw_items:
                 self._check_cancelled()
-                title = item.get("title")
+                title = str(item.get('title') or '').strip()
                 if not title or not self._is_english_dominant(title):
                     continue
-                year = item.get("publication_year", "n.d.")
+                year = item.get('publication_year') or 'n.d.'
                 formatted_authors, last_names = [], []
-                for authorship in item.get("authorships", []):
-                    name_raw = authorship.get("author", {}).get("display_name", "")
-                    if not name_raw or not name_raw.strip():
+                for authorship in item.get('authorships') or []:
+                    name = str((authorship.get('author') or {}).get('display_name') or '').strip()
+                    if not name:
                         continue
-                    parts = name_raw.strip().split()
-                    if len(parts) > 1:
-                        last_names.append(parts[-1])
-                        formatted_authors.append(f"{parts[-1]}, {parts[0][0]}.")
-                    else:
-                        last_names.append(name_raw)
-                        formatted_authors.append(name_raw)
+                    parts = name.split(); last = parts[-1]
+                    last_names.append(last)
+                    formatted_authors.append(f"{last}, {parts[0][0]}." if len(parts) > 1 else last)
                 if not formatted_authors:
                     continue
                 if len(last_names) == 1:
@@ -1523,170 +1760,175 @@ Return ONLY a valid JSON array of strings."""
                     in_text = f"({last_names[0]} & {last_names[1]}, {year})"
                 else:
                     in_text = f"({last_names[0]} et al., {year})"
-                if len(formatted_authors) > 2:
-                    author_str = ", ".join(formatted_authors[:-1]) + ", & " + formatted_authors[-1]
-                elif len(formatted_authors) == 2:
-                    author_str = " & ".join(formatted_authors)
-                else:
-                    author_str = formatted_authors[0]
-                doi_clean = (item.get("doi") or "").replace("https://doi.org/", "")
-                venue_display = ((item.get("primary_location") or {}).get("source") or {}).get("display_name", "Journal")
-                apa_full = f"{author_str} ({year}). {title}. {venue_display}."
-                if doi_clean:
-                    apa_full += f" https://doi.org/{doi_clean}"
-                raw_abstract = " ".join([
-                    word
-                    for position, word in sorted([
-                        (position, word)
-                        for word, positions in (item.get("abstract_inverted_index") or {}).items()
-                        for position in positions
-                    ])
-                ])
-                abstract_str = " ".join(str(raw_abstract).split()[:60])
+                author_str = (', '.join(formatted_authors[:-1]) + ', & ' + formatted_authors[-1]) if len(formatted_authors) > 2 else (' & '.join(formatted_authors) if len(formatted_authors) == 2 else formatted_authors[0])
+                doi = self._normalize_doi(item.get('doi'))
+                venue = (((item.get('primary_location') or {}).get('source') or {}).get('display_name') or 'Journal')
+                ref = f"{author_str} ({year}). {title}. {venue}."
+                if doi:
+                    ref += f" https://doi.org/{doi}"
+                inverted = item.get('abstract_inverted_index') or {}
+                words = [word for position, word in sorted((position, word) for word, positions in inverted.items() for position in positions)]
+                abstract = ' '.join(words[:180])
                 results.append({
-                    "id": (title.lower().strip(), str(year)),
-                    "in_text": in_text,
-                    "ref": apa_full,
-                    "rag_text": f"Title: {title}. Focus: {abstract_str}...",
-                    "source": "oa",
+                    'title': title,
+                    'year': str(year),
+                    'doi': doi,
+                    'external_id': str(item.get('id') or ''),
+                    'in_text': in_text,
+                    'ref': ref,
+                    'rag_text': f"Title: {title}. Abstract evidence: {abstract}",
+                    'source': 'oa',
                 })
+            self._source_metric_add('oa', 'eligible_result_count', len(results))
+            self._source_metric_set('oa', 'status', 'ok' if results else 'no_results')
+            if not results:
+                self._source_metric_set('oa', 'last_safe_error', 'OPENALEX_NO_RESULTS')
+            return results
         except PipelineCancelledError:
             raise
+        except requests.Timeout as error:
+            self._record_source_error('oa', 'OPENALEX_TIMEOUT', detail=error)
         except Exception as error:
-            self._source_metric_add("oa", "error_count")
-            status = getattr(getattr(error, "response", None), "status_code", None)
-            if status is not None:
-                self._source_metric_set("oa", "last_status_code", status)
-            safe_error = self._safe_source_error(error)
-            self._source_metric_set("oa", "last_safe_error", safe_error)
-            logging.warning("OpenAlex search error: %s", safe_error)
-        return results
+            status = getattr(getattr(error, 'response', None), 'status_code', None)
+            if status == 429:
+                code = 'OPENALEX_RATE_LIMIT'
+            elif status in (401, 403):
+                code = 'OPENALEX_AUTH_ERROR'
+            elif status == 400:
+                code = 'OPENALEX_BAD_QUERY'
+            else:
+                code = 'OPENALEX_REQUEST_ERROR'
+            self._record_source_error('oa', code, status, error)
+        return []
 
     def fetch_references_parallel(self, topic, keywords, min_required=50, progress_callback=None):
+        """Retrieve adaptively, then merge deterministically by DOI/ID/title-year.
+
+        Each phase first requests one page per source and query. A second page is
+        requested only when the minimum evidence set is still not met. This keeps
+        the interface responsive and avoids redundant OpenAlex filter pairs.
+        """
         with self.lock:
             self.citation_map.clear()
             self.used_citations.clear()
-            for source in ("s2", "oa"):
+            self.citation_audit_log.clear()
+            for source in ('s2', 'oa'):
                 for field in (
-                    "request_count", "success_count", "raw_result_count",
-                    "unique_added_count", "duplicate_count", "error_count",
+                    'request_count','success_count','raw_result_count',
+                    'eligible_result_count','merged_count','unique_added_count',
+                    'duplicate_count','error_count',
                 ):
                     self.source_metrics[source][field] = 0
-                self.source_metrics[source]["last_status_code"] = None
-                self.source_metrics[source]["last_safe_error"] = ""
-            self.s2_calls = 0
-            self.oa_calls = 0
+                self.source_metrics[source]['last_status_code'] = None
+                self.source_metrics[source]['last_safe_error'] = ''
+                self.source_metrics[source]['status'] = 'not_started'
+            self.s2_calls = self.oa_calls = 0
 
-        llm_rag_context = []
-        seen_identifiers = {}
         all_used_terms = set()
-        filters_to_try = [
-            "language:en,type:article,has_doi:true,is_paratext:false",
-            "language:en,type:article,is_paratext:false",
-        ]
+        collected = {'s2': [], 'oa': []}
+        basic_filter = 'language:en,type:article,is_paratext:false'
+
+        def unique_preview_count():
+            keys_seen = set()
+            count = 0
+            for source in ('s2', 'oa'):
+                for item in collected[source]:
+                    keys = self._reference_keys(item)
+                    if keys and any(key in keys_seen for key in keys):
+                        continue
+                    keys_seen.update(keys)
+                    count += 1
+            return count
 
-        for expansion_phase in range(3):
+        def run_wave(terms, *, second_page=False):
+            future_sources = {}
+            with concurrent.futures.ThreadPoolExecutor(max_workers=self.lit_max_workers) as executor:
+                if not self.s2_fused:
+                    offset = 50 if second_page else 0
+                    for term in terms:
+                        future_sources[executor.submit(self._s2_worker, term, offset)] = 's2'
+                if not self.oa_fused:
+                    page = 2 if second_page else 1
+                    for term in terms:
+                        future_sources[executor.submit(self._oa_worker, term, page, basic_filter)] = 'oa'
+                for future in concurrent.futures.as_completed(future_sources):
+                    self._check_cancelled()
+                    source = future_sources[future]
+                    collected[source].extend(future.result())
+                    if progress_callback:
+                        progress_callback()
+
+        for phase in range(3):
             self._check_cancelled()
-            if len(self.citation_map) >= min_required:
-                break
-            if expansion_phase == 0:
-                search_terms = self.generate_search_queries(topic, keywords)
+            if phase == 0:
+                terms = self.generate_search_queries(topic, keywords)
             else:
-                system_prompt = """You are an expert academic librarian. Broaden the search scope. Return only a JSON array containing three new English search queries, each under five terms. Do not use nested Boolean logic."""
-                user_prompt = (
-                    f"Topic: {topic}\nKeywords: {keywords}\n"
-                    f"Used Queries: {list(all_used_terms)}"
-                )
                 response = self.call_deepseek(
-                    system_prompt,
-                    user_prompt,
-                    temp=0.5,
-                    model_role="fast",
+                    'You are an academic librarian. Return only a JSON array with three new concise English search queries, each under five terms.',
+                    f"Topic: {topic}\nKeywords: {keywords}\nUsed: {sorted(all_used_terms)}",
+                    temp=0.4,
+                    model_role='fast',
                 )
-                queries = self._extract_json_from_text(response)
-                if isinstance(queries, list):
-                    search_terms = [
-                        str(query).strip()
-                        for query in queries
-                        if self._is_english_dominant(str(query).strip())
-                    ][:3]
-                else:
-                    search_terms = []
-            current_batch_terms = [
-                term for term in search_terms if term not in all_used_terms
-            ]
-            if not current_batch_terms:
+                parsed = self._extract_json_from_text(response)
+                terms = parsed if isinstance(parsed, list) else []
+
+            normalized_terms = []
+            for value in terms:
+                term = self._normalize_search_query(value)
+                if term and self._is_english_dominant(term) and term not in all_used_terms:
+                    normalized_terms.append(term)
+                    all_used_terms.add(term)
+            if not normalized_terms:
                 continue
-            all_used_terms.update(current_batch_terms)
 
-            executor = concurrent.futures.ThreadPoolExecutor(
-                max_workers=self.lit_max_workers
-            )
-            futures = []
-            try:
-                if not self.s2_fused:
-                    for term in current_batch_terms:
-                        futures.append(executor.submit(self._s2_worker, term, 0))
-                        futures.append(executor.submit(self._s2_worker, term, 50))
-                if not self.oa_fused:
-                    for term in current_batch_terms:
-                        for filter_str in filters_to_try:
-                            futures.append(executor.submit(self._oa_worker, term, 1, filter_str))
-                            futures.append(executor.submit(self._oa_worker, term, 2, filter_str))
-
-                for future in concurrent.futures.as_completed(futures):
-                    self._check_cancelled()
-                    results = future.result()
-                    with self.lock:
-                        for result in results:
-                            source = result["source"]
-                            identifier = result["id"]
-                            if identifier in seen_identifiers:
-                                self.source_metrics[source]["duplicate_count"] += 1
-                                existing_tag = seen_identifiers[identifier]
-                                sources = self.citation_map[existing_tag].setdefault("sources", [])
-                                if source not in sources:
-                                    sources.append(source)
-                                continue
-                            if len(self.citation_map) >= self.lit_target_papers:
-                                break
-                            seen_identifiers[identifier] = f"[CITE-{len(self.citation_map)+1:02d}]"
-                            cite_tag = seen_identifiers[identifier]
-                            self.citation_map[cite_tag] = {
-                                "in_text": result["in_text"],
-                                "ref": result["ref"],
-                                "id": identifier,
-                                "evidence": result.get("rag_text", ""),
-                                "sources": [source],
-                            }
-                            self.source_metrics[source]["unique_added_count"] += 1
-                            llm_rag_context.append(
-                                f"{cite_tag} {result['rag_text']}"
-                            )
-                        self.s2_calls = self.source_metrics["s2"]["unique_added_count"]
-                        self.oa_calls = self.source_metrics["oa"]["unique_added_count"]
-                    if progress_callback:
-                        progress_callback()
-                    if len(self.citation_map) >= self.lit_target_papers:
-                        break
-            except PipelineCancelledError:
-                for future in futures:
-                    future.cancel()
-                # Do not wait for unrelated network workers after the user has
-                # explicitly cancelled or logged out.
-                executor.shutdown(wait=False, cancel_futures=True)
-                executor = None
-                raise
-            finally:
-                if executor is not None:
-                    executor.shutdown(wait=True, cancel_futures=False)
+            run_wave(normalized_terms, second_page=False)
+            if unique_preview_count() >= min_required:
+                break
+            run_wave(normalized_terms, second_page=True)
+            if unique_preview_count() >= min_required:
+                break
 
+        merge_limit = max(min_required, min(self.lit_target_papers, min_required + 30))
+        merged = self._merge_reference_results(collected, merge_limit)
+        with self.lock:
+            for index, result in enumerate(merged, start=1):
+                tag = f"[CITE-{index:02d}]"
+                result_keys = self._reference_keys(result)
+                self.citation_map[tag] = {
+                    'in_text': result['in_text'],
+                    'ref': result['ref'],
+                    'id': result_keys[0] if result_keys else ('title_year', result.get('title'), result.get('year')),
+                    'doi': result.get('doi',''),
+                    'external_id': result.get('external_id',''),
+                    'title': result.get('title',''),
+                    'year': result.get('year',''),
+                    'evidence': result.get('rag_text',''),
+                    'sources': list(result.get('sources') or [result.get('source')]),
+                }
+            self.s2_calls = int(self.source_metrics['s2']['eligible_result_count'])
+            self.oa_calls = int(self.source_metrics['oa']['eligible_result_count'])
+            for source, no_result_code in (('s2', 'S2_NO_RESULTS'), ('oa', 'OPENALEX_NO_RESULTS')):
+                metric = self.source_metrics[source]
+                eligible = int(metric.get('eligible_result_count', 0))
+                errors = int(metric.get('error_count', 0))
+                if eligible > 0 and errors > 0:
+                    metric['status'] = 'partial'
+                elif eligible > 0:
+                    metric['status'] = 'ok'
+                    metric['last_safe_error'] = ''
+                elif errors > 0:
+                    metric['status'] = 'error'
+                else:
+                    metric['status'] = 'no_results'
+                    metric['last_safe_error'] = no_result_code
         logging.info(
-            "Reference search completed: unique=%s metrics=%s",
+            'Reference search completed: unique=%s metrics=%s',
             len(self.citation_map),
             self.get_source_metrics(),
         )
-        return "\n".join(llm_rag_context)
+        return '\n'.join(
+            f"{tag} {meta['evidence']}" for tag, meta in self.citation_map.items()
+        )
 
     @staticmethod
     def _citation_tag_number(tag):
@@ -1819,12 +2061,13 @@ Return ONLY a valid JSON array of strings."""
         text,
         section_name="",
     ):
-        """Audit cited sentences against retrieved evidence and fail closed.
+        """Audit cited sentences against retrieved evidence at sentence scope.
 
         The audit may split a sentence when different sources support different
         claims, or keep one source when several sources support the same claim.
         It may not invent facts, numbers, quotations, or citation tags. A cited
-        claim that cannot be supported by the available evidence stops the task.
+        claim that cannot be supported is removed from the manuscript and logged;
+        one unsupported sentence must not discard an otherwise valid paper.
         """
         if not text:
             return text
@@ -1945,8 +2188,6 @@ Return exactly:
                         self._safe_source_error(error),
                     )
             if last_error is not None:
-                if not fail_closed:
-                    continue
                 raise CitationAuditError(
                     "E_CITATION_AUDIT_TIMEOUT",
                     self._safe_source_error(last_error),
@@ -2013,13 +2254,25 @@ Return exactly:
                         f"Invalid citation audit status for {record['id']}: {status!r}"
                     )
                 if status == "unsupported":
+                    reason = str(item.get("reason", ""))[:500]
                     unsupported.append(
                         {
                             "id": record["id"],
                             "sentence": record["original"][:500],
-                            "reason": str(item.get("reason", ""))[:500],
+                            "reason": reason,
                         }
                     )
+                    # An unsupported externally verifiable claim cannot remain in
+                    # the manuscript with an arbitrary citation. Remove only this
+                    # sentence and retain an auditable record of the decision.
+                    replacements[record["id"]] = ""
+                    self._append_citation_audit(
+                        section_name,
+                        "remove_unsupported_claim",
+                        record["original"],
+                        "",
+                        reason or "No candidate evidence directly supported the claim.",
+                    )
                     continue
                 if not revised:
                     raise DeepSeekRequestError(
@@ -2056,12 +2309,11 @@ Return exactly:
                 replacements[record["id"]] = revised
 
         if unsupported:
-            raise CitationAuditError(
-                "E_CITATION_UNSUPPORTED",
-                "Unsupported cited claims: " + json.dumps(
-                    unsupported[:3],
-                    ensure_ascii=False,
-                ),
+            logging.warning(
+                "Citation audit removed unsupported claims: section=%s count=%s examples=%s",
+                section_name,
+                len(unsupported),
+                json.dumps(unsupported[:3], ensure_ascii=False),
             )
 
         rebuilt = []
@@ -2072,6 +2324,8 @@ Return exactly:
             cursor = record["end"]
         rebuilt.append(str(text)[cursor:])
         audited = "".join(rebuilt)
+        audited = re.sub(r"[ \t]+\n", "\n", audited)
+        audited = re.sub(r"\n{3,}", "\n\n", audited).strip()
         violations = self.find_multi_citation_violations(audited)
         if violations:
             raise DeepSeekRequestError(
@@ -2085,6 +2339,598 @@ Return exactly:
         )
         return audited
 
+    def _append_citation_audit(self, section_name, action, original, revised='', detail=''):
+        self.citation_audit_log.append({
+            'section': str(section_name),
+            'action': str(action),
+            'original': str(original)[:800],
+            'revised': str(revised)[:800],
+            'detail': str(detail)[:500],
+        })
+
+    def _citation_support_score(self, sentence, tag):
+        metadata = self.citation_map.get(tag) or {}
+        claim_text = re.sub(r"\[?CITE-\d+\]?", " ", str(sentence), flags=re.I)
+        claim_terms = self._citation_terms(claim_text)
+        evidence_text = " ".join(
+            [str(metadata.get("ref", "")), str(metadata.get("evidence", ""))]
+        )
+        evidence_terms = self._citation_terms(evidence_text)
+        overlap = len(claim_terms & evidence_terms)
+        coverage = overlap / max(1, len(claim_terms))
+        return coverage, overlap
+
+    def _local_citation_recovery(self, text, section_name=''):
+        """Conservative deterministic recovery for technical audit failures.
+
+        This routine never keeps the first citation merely because it appeared
+        first. A source is retained only when lexical evidence is sufficiently
+        strong; otherwise the externally verifiable sentence is removed and the
+        decision is written to the citation audit log.
+        """
+        rebuilt = []
+        cursor = 0
+        for start, end, sentence in self._sentence_spans(text):
+            rebuilt.append(str(text)[cursor:start])
+            original = sentence
+            tags = [f"[{tag.upper()}]" for tag in re.findall(r'CITE-\d+', sentence, re.I)]
+            unique_tags = list(dict.fromkeys(tags))
+            valid = [tag for tag in unique_tags if tag in self.citation_map]
+            revised = sentence
+
+            if unique_tags and not valid:
+                revised = re.sub(r'\s*[\[(]?CITE-\d+[\])]?\s*', ' ', sentence, flags=re.I)
+                revised = re.sub(r'\s+', ' ', revised).strip()
+                self._append_citation_audit(
+                    section_name,
+                    'remove_unknown_tag',
+                    original,
+                    revised,
+                    'No valid citation tag was available.',
+                )
+            elif len(valid) == 1:
+                selected_tag = valid[0]
+                coverage, overlap = self._citation_support_score(sentence, selected_tag)
+                if overlap >= 2 and coverage >= 0.12:
+                    self._append_citation_audit(
+                        section_name,
+                        'retain_lexically_supported_source',
+                        original,
+                        sentence,
+                        f'Lexical support overlap={overlap}, coverage={coverage:.3f}.',
+                    )
+                else:
+                    revised = ''
+                    self._append_citation_audit(
+                        section_name,
+                        'remove_unverified_claim',
+                        original,
+                        '',
+                        f'The single candidate did not meet the conservative support threshold; overlap={overlap}, coverage={coverage:.3f}.',
+                    )
+            elif len(valid) > 1:
+                scored = sorted(
+                    (
+                        (*self._citation_support_score(sentence, tag), -self._citation_tag_number(tag), tag)
+                        for tag in valid
+                    ),
+                    reverse=True,
+                )
+                coverage, overlap, _, selected_tag = scored[0]
+                if overlap >= 2 and coverage >= 0.12:
+                    revised = re.sub(
+                        r'\s*[\[(]?CITE-\d+(?:\s*[,;]\s*CITE-\d+)*[\])]?\s*',
+                        ' ',
+                        sentence,
+                        flags=re.I,
+                    )
+                    revised = re.sub(r'\s+', ' ', revised).strip()
+                    punctuation = revised[-1] if revised and revised[-1] in '.!?。！？' else ''
+                    if punctuation:
+                        revised = revised[:-1].rstrip()
+                    revised = f"{revised} {selected_tag}{punctuation}"
+                    self._append_citation_audit(
+                        section_name,
+                        'select_evidence_supported_source',
+                        original,
+                        revised,
+                        f'Lexical support overlap={overlap}, coverage={coverage:.3f}.',
+                    )
+                else:
+                    revised = ''
+                    self._append_citation_audit(
+                        section_name,
+                        'remove_unverified_claim',
+                        original,
+                        '',
+                        f'No candidate met the conservative support threshold; best overlap={overlap}, coverage={coverage:.3f}.',
+                    )
+            rebuilt.append(revised)
+            cursor = end
+        rebuilt.append(str(text)[cursor:])
+        recovered = ''.join(rebuilt)
+        recovered = re.sub(r'[ \t]+\n', '\n', recovered)
+        return re.sub(r'\n{3,}', '\n\n', recovered).strip()
+
+    def audit_and_resolve_citations(self, text, section_name='', quarantine=False):
+        value = str(text or '')
+        if quarantine:
+            return self.resolve_citations(value, quarantine=True)
+        unknown = sorted({f"[{tag.upper()}]" for tag in re.findall(r'CITE-\d+', value, re.I)} - set(self.citation_map))
+        if unknown:
+            for tag in unknown:
+                value = re.sub(rf'\s*\[?{re.escape(tag.strip("[]"))}\]?\s*', ' ', value, flags=re.I)
+            self._append_citation_audit(section_name, 'remove_unknown_tags', text, value, ','.join(unknown))
+        has_citations = bool(re.search(r'CITE-\d+', value, re.I))
+        violations = self.find_multi_citation_violations(value)
+        if has_citations and self.citation_semantic_audit.get('enabled', True):
+            try:
+                value = self.semantic_audit_claim_specific_citations(value, section_name=section_name)
+            except PipelineCancelledError:
+                raise
+            except CitationAuditError as error:
+                logging.warning(
+                    'Citation semantic audit technical fallback: section=%s code=%s detail=%s',
+                    section_name,
+                    error.code,
+                    error.safe_detail,
+                )
+                value = self._local_citation_recovery(value, section_name)
+            except Exception as error:
+                logging.warning(
+                    'Citation semantic audit fallback: section=%s error=%s',
+                    section_name,
+                    self._safe_source_error(error),
+                )
+                value = self._local_citation_recovery(value, section_name)
+        elif violations:
+            value = self._local_citation_recovery(value, section_name)
+        if self.find_multi_citation_violations(value):
+            value = self._local_citation_recovery(value, section_name)
+        if self.find_multi_citation_violations(value):
+            raise DeepSeekRequestError(
+                f'Citation recovery could not resolve all violations in {section_name}.'
+            )
+        return self.resolve_citations(value, quarantine=False)
+
+    @staticmethod
+    def _remove_ai_meta_traces(text):
+        patterns = [
+            r'(?im)^.*\bas an ai(?: language model)?\b.*(?:\n|$)',
+            r'(?im)^.*\bai[- ]generated\b.*(?:\n|$)',
+            r'(?im)^.*\bthis response\b.*(?:\n|$)',
+            r'(?im)^.*\bi cannot (?:browse|access|verify)\b.*(?:\n|$)',
+        ]
+        value = str(text or '')
+        for pattern in patterns:
+            value = re.sub(pattern, '', value)
+        return re.sub(r'\n{3,}', '\n\n', value).strip()
+
+    def polish_academic_section(self, text, section_name='', abstract=False):
+        value = self._remove_ai_meta_traces(text)
+        quality = APP_CONFIG.get('manuscript_quality', {})
+        forbidden = quality.get('forbidden_phrases', [])
+        detected = [phrase for phrase in forbidden if re.search(rf'\b{re.escape(phrase)}\b', value, re.I)]
+        transitions = re.findall(r'(?im)^(Firstly|Secondly|Moreover|Furthermore|In conclusion|To summarize)\b', value)
+        numeric_count = len(re.findall(r'(?<![A-Za-z])[-+]?\d+(?:\.\d+)?%?', value)) if abstract else 0
+        max_numbers = int(quality.get('abstract_max_numeric_mentions', 6))
+        needs_revision = bool(detected or transitions or (abstract and numeric_count > max_numbers))
+        if not needs_revision:
+            return value
+        numeric_rule = (
+            "For the abstract, remove nonessential numerical details. You may omit numbers, "
+            "but do not alter any number that remains and do not add a new number."
+            if abstract
+            else "Preserve every number, percentage, coefficient, p-value, and confidence interval exactly."
+        )
+        prompt = f"""Revise only the supplied academic section. Preserve every heading, citation tag, quotation, hypothesis label, variable abbreviation, and conclusion direction exactly. {numeric_rule} Use concise, accurate, rigorous academic English. Remove formulaic transitions, repetitive summaries, metaphors, analogies, promotional language, and any AI-writing meta-commentary. Do not add new evidence or citations.\nSECTION: {section_name}\nTEXT:\n{value}"""
+        revised = self.call_deepseek(
+            'You are a meticulous academic language editor. Output only the revised section.',
+            prompt,
+            max_tokens=max(1200, min(8000, self._count_tokens(value) + 1200)),
+            temp=0.1,
+            model_role='writing',
+            thinking=False,
+            allow_fallback=False,
+        )
+        revised = self._remove_ai_meta_traces(revised)
+        original_numbers = self._numeric_tokens(value)
+        revised_numbers = self._numeric_tokens(revised)
+        if abstract:
+            added_or_changed = any(
+                count > original_numbers.get(token, 0)
+                for token, count in revised_numbers.items()
+            )
+            if added_or_changed:
+                logging.warning('Abstract quality revision rejected because a number was added or changed: %s', section_name)
+                return value
+            if sum(revised_numbers.values()) > max_numbers:
+                logging.warning('Abstract quality revision rejected because numerical density remains excessive: %s', section_name)
+                return value
+        elif original_numbers != revised_numbers:
+            logging.warning('Quality revision rejected because numbers changed: %s', section_name)
+            return value
+        original_tags = Counter(re.findall(r'CITE-\d+', value, re.I))
+        revised_tags = Counter(re.findall(r'CITE-\d+', revised, re.I))
+        if original_tags != revised_tags:
+            logging.warning('Quality revision rejected because citation tags changed: %s', section_name)
+            return value
+        return revised.strip()
+
+    @staticmethod
+    def _trim_at_sentence_boundary(value, max_words):
+        text = str(value or '').strip()
+        words = text.split()
+        if len(words) <= max_words:
+            return text
+        sentences = re.split(r'(?<=[.!?])\s+', text)
+        selected = []
+        count = 0
+        for sentence in sentences:
+            sentence_words = sentence.split()
+            if not sentence_words:
+                continue
+            if count + len(sentence_words) > max_words:
+                break
+            selected.append(sentence.strip())
+            count += len(sentence_words)
+        if selected:
+            return ' '.join(selected).strip()
+        return ' '.join(words[:max_words]).rstrip(' ,;:') + '.'
+
+    def _abstract_manuscript_context(self, manuscript, max_tokens=18000):
+        value = str(manuscript or '').strip()
+        if self._count_tokens(value) <= max_tokens:
+            return value
+        matches = list(re.finditer(r'(?m)^#\s+(.+?)\s*$', value))
+        if not matches:
+            head = self._truncate_context(value, max_tokens // 3)
+            tail_tokens = self.tokenizer.encode(value)[-(max_tokens - self._count_tokens(head)):]
+            return head + '\n\n' + self.tokenizer.decode(tail_tokens)
+
+        sections = []
+        preamble = value[:matches[0].start()].strip()
+        if preamble:
+            sections.append(('preamble', preamble))
+        for index, match in enumerate(matches):
+            end = matches[index + 1].start() if index + 1 < len(matches) else len(value)
+            sections.append((match.group(1).strip().lower(), value[match.start():end].strip()))
+
+        categories = (
+            (('method', 'research design', 'data', 'sample'), 3000),
+            (('result', 'finding', 'analysis'), 6500),
+            (('discussion', 'implication'), 4000),
+            (('conclusion',), 2500),
+            (('introduction', 'background'), 1800),
+        )
+        selected = []
+        used = set()
+        for keywords, budget in categories:
+            candidates = [
+                (index, text) for index, (title, text) in enumerate(sections)
+                if index not in used and any(keyword in title for keyword in keywords)
+            ]
+            for index, text in candidates:
+                selected.append(self._truncate_context(text, budget))
+                used.add(index)
+                break
+        if not selected:
+            return self._truncate_context(value, max_tokens)
+        combined = '\n\n'.join(selected)
+        return self._truncate_context(combined, max_tokens)
+
+    def generate_final_abstract(self, topic, keywords, manuscript, data_evidence='', mode='amj'):
+        evidence = self._truncate_context(str(data_evidence or ''), 5000)
+        body = self._abstract_manuscript_context(manuscript, 18000)
+        prompt = f"""Write the final English abstract for this completed {mode.upper()} manuscript. Maximum 200 words. State the research problem, design, main finding, and contribution. Use no citations. Include only the minimum numerical evidence required; do not list coefficients, p-values, and fit statistics unless one value is indispensable. Use concise, accurate, rigorous academic language. Do not use metaphors, analogies, rhetorical questions, promotional language, formulaic transitions, or AI-related meta-commentary. Output only the abstract.\nTOPIC: {topic}\nKEYWORDS: {keywords}\nFROZEN EVIDENCE:\n{evidence}\nCOMPLETED MANUSCRIPT:\n{body}"""
+        abstract = self.call_deepseek(
+            'You are a senior journal editor.', prompt, max_tokens=900, temp=0.1,
+            model_role='writing', thinking=False, allow_fallback=False,
+        )
+        abstract = self._remove_ai_meta_traces(abstract)
+        abstract = re.sub(r'(?im)^#+\s*abstract\s*', '', abstract).strip()
+        abstract = self.polish_academic_section(abstract, 'Abstract', abstract=True)
+        max_words = max(80, int(APP_CONFIG.get('manuscript_quality', {}).get('abstract_max_words', 200)))
+        return self._trim_at_sentence_boundary(abstract, max_words)
+
+    @staticmethod
+    def _translation_language_rule(target_lang):
+        if 'Russian' in target_lang:
+            return 'Use idiomatic academic Russian with correct case government, gender-number agreement, aspect, punctuation, and terminology. Reorganize English syntax where necessary and avoid literal English calques.'
+        if 'Kazakh' in target_lang:
+            return 'Use idiomatic academic Kazakh with correct case and possessive suffixes, person/plural endings, vowel harmony, consonant assimilation, natural word order, and correct Kazakh-specific letters. Avoid Russian syntactic calques.'
+        return 'Use formal academic Simplified Chinese with explicit concepts, natural Chinese syntax, clear hierarchy, and logically connected claims. Avoid carrying English nominalization and clause order into Chinese mechanically.'
+
+    @staticmethod
+    def _translation_prompt(target_lang):
+        common = """Translate the supplied blocks from the locked English manuscript faithfully. Preserve every marker, citation, number, percentage, coefficient, p-value, confidence interval, DOI, hypothesis label, variable abbreviation, quotation meaning, and conclusion direction. Do not add or remove claims. Use concise, accurate, rigorous academic language. Avoid metaphors, analogies, rhetorical questions, promotional language, formulaic transitions, and mechanical repetition. Return every marker exactly once and in the same order. Output only the marked translations."""
+        return common + ' ' + SSciPipelineAgent._translation_language_rule(target_lang)
+
+    @staticmethod
+    def _localized_reference_heading(target_lang):
+        if 'Russian' in target_lang:
+            return 'Литература'
+        if 'Kazakh' in target_lang:
+            return 'Пайдаланылған әдебиеттер'
+        return '参考文献'
+
+    @staticmethod
+    def _markdown_heading_tree(value):
+        return [
+            (len(match.group(1)), match.group(2).strip())
+            for match in re.finditer(r'(?m)^(#{1,3})\s+(.+?)\s*$', str(value or ''))
+        ]
+
+    @staticmethod
+    def _translation_hypothesis_tokens(value):
+        return Counter(re.findall(r'\bH\d+[A-Za-z]?\b', str(value or ''), flags=re.I))
+
+    @staticmethod
+    def _translation_acronym_tokens(value):
+        without_doi = re.sub(r'https?://\S+|doi:\s*\S+', ' ', str(value or ''), flags=re.I)
+        return Counter(re.findall(r'\b[A-Z][A-Z0-9-]{1,11}\b', without_doi))
+
+    @staticmethod
+    def _translation_citation_tokens(value):
+        return Counter(
+            re.findall(
+                r'\([^()\n]*?(?:19|20)\d{2}[a-z]?[^()\n]*?\)',
+                str(value or ''),
+                flags=re.I,
+            )
+        )
+
+    @staticmethod
+    def _translation_body_units(body):
+        units = []
+        buffer = []
+
+        def flush_buffer():
+            if not buffer:
+                return
+            segment = '\n'.join(buffer).strip()
+            buffer.clear()
+            if not segment:
+                return
+            for part in re.split(r'\n{2,}', segment):
+                part = part.strip()
+                if part:
+                    units.append({'kind': 'body', 'text': part})
+
+        for line in str(body or '').splitlines():
+            heading = re.match(r'^(#{1,3})\s+(.+?)\s*$', line)
+            if heading:
+                flush_buffer()
+                units.append({
+                    'kind': 'heading',
+                    'level': len(heading.group(1)),
+                    'text': heading.group(2).strip(),
+                })
+            else:
+                buffer.append(line)
+        flush_buffer()
+        return units
+
+    def _build_translation_glossary(self, body, target_lang):
+        headings = [title for _, title in self._markdown_heading_tree(body)]
+        acronyms = sorted(self._translation_acronym_tokens(body))
+        phrases = re.findall(
+            r'\b(?:[A-Z][a-z]+(?:[- ][A-Z][a-z]+){1,3})\b',
+            str(body or ''),
+        )
+        phrase_counts = Counter(phrases)
+        term_limit = max(10, int(APP_CONFIG.get('translation_glossary_term_limit', 40)))
+        recurring = [term for term, count in phrase_counts.most_common() if count >= 2]
+        unique_headings = list(dict.fromkeys(headings))
+        extras = [term for term in list(dict.fromkeys(recurring + acronyms)) if term not in unique_headings]
+        sources = unique_headings + extras[:max(0, term_limit - len(unique_headings))]
+        if not sources:
+            return {}
+
+        language_rule = self._translation_language_rule(target_lang)
+        payload = json.dumps({'terms': sources}, ensure_ascii=False)
+        system_prompt = f"""Create one consistent academic terminology glossary for translation into {target_lang}. {language_rule}
+Return valid JSON only in this exact form: {{"terms":[{{"source":"...","target":"..."}}]}}.
+Translate headings and terminology naturally. Preserve standard variable abbreviations and acronyms when they should remain unchanged. Return every supplied source exactly once."""
+        response = self.call_deepseek(
+            system_prompt,
+            payload,
+            max_tokens=max(1600, min(6000, len(sources) * 80)),
+            temp=0.0,
+            model_role='translation',
+            thinking=False,
+            allow_fallback=False,
+        )
+        parsed = self._extract_json_from_text(response)
+        items = parsed.get('terms', []) if isinstance(parsed, dict) else []
+        mapping = {
+            str(item.get('source') or '').strip(): str(item.get('target') or '').strip()
+            for item in items
+            if isinstance(item, dict)
+        }
+        missing_headings = [heading for heading in headings if not mapping.get(heading)]
+        if missing_headings:
+            raise DeepSeekRequestError(
+                'Translation glossary omitted heading(s): ' + ', '.join(missing_headings[:5])
+            )
+        for acronym in acronyms:
+            mapping.setdefault(acronym, acronym)
+        return mapping
+
+    @staticmethod
+    def _parse_marked_translation(value, expected_ids):
+        pattern = re.compile(r'(?m)^\[\[SP-P(\d{6})\]\]\s*$')
+        matches = list(pattern.finditer(str(value or '')))
+        found_ids = [match.group(1) for match in matches]
+        if found_ids != list(expected_ids):
+            raise DeepSeekRequestError(
+                f'Translation marker mismatch: expected={list(expected_ids)}, got={found_ids}.'
+            )
+        parsed = {}
+        for index, match in enumerate(matches):
+            end = matches[index + 1].start() if index + 1 < len(matches) else len(str(value or ''))
+            translated = str(value or '')[match.end():end].strip()
+            if not translated:
+                raise DeepSeekRequestError(f'Translation block SP-P{match.group(1)} is empty.')
+            parsed[match.group(1)] = translated
+        return parsed
+
+    def _validate_translation_block(self, source, translated, block_id, target_lang):
+        if self._numeric_tokens(source) != self._numeric_tokens(translated):
+            raise DeepSeekRequestError(f'Translation changed numerical content in {block_id}.')
+        if self._translation_citation_tokens(source) != self._translation_citation_tokens(translated):
+            raise DeepSeekRequestError(f'Translation changed citations in {block_id}.')
+        if self._translation_hypothesis_tokens(source) != self._translation_hypothesis_tokens(translated):
+            raise DeepSeekRequestError(f'Translation changed hypothesis labels in {block_id}.')
+        if self._translation_acronym_tokens(source) != self._translation_acronym_tokens(translated):
+            raise DeepSeekRequestError(f'Translation changed variable abbreviations in {block_id}.')
+
+        source_compact = re.sub(r'\s+', '', str(source or ''))
+        target_compact = re.sub(r'\s+', '', str(translated or ''))
+        if len(source_compact) >= 120:
+            is_chinese = 'Chinese' in target_lang
+            minimum = float(APP_CONFIG.get(
+                'translation_min_block_ratio_zh' if is_chinese else 'translation_min_block_ratio_cyrillic',
+                0.28 if is_chinese else 0.65,
+            ))
+            maximum = float(APP_CONFIG.get('translation_max_block_ratio', 2.20))
+            ratio = len(target_compact) / max(1, len(source_compact))
+            if ratio < minimum or ratio > maximum:
+                raise DeepSeekRequestError(
+                    f'Translation coverage ratio is implausible in {block_id}: {ratio:.3f}.'
+                )
+            if source_compact.casefold() == target_compact.casefold():
+                raise DeepSeekRequestError(f'Translation block {block_id} was returned unchanged.')
+            if is_chinese and not re.search(r'[\u4e00-\u9fff]', translated):
+                raise DeepSeekRequestError(f'Translation block {block_id} contains no Chinese text.')
+            if not is_chinese and not re.search(r'[А-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі]', translated):
+                raise DeepSeekRequestError(f'Translation block {block_id} contains no Cyrillic target-language text.')
+
+    def translate_locked_manuscript(self, manuscript, target_lang):
+        value = str(manuscript or '').strip()
+        if not value:
+            raise DeepSeekRequestError('English manuscript is empty.')
+        reference_split = re.split(r'(?m)^#\s+References\s*$', value, maxsplit=1)
+        body = reference_split[0].strip()
+        references = reference_split[1].strip() if len(reference_split) == 2 else ''
+        units = self._translation_body_units(body)
+        if not units:
+            raise DeepSeekRequestError('English manuscript contains no translatable content.')
+
+        glossary = self._build_translation_glossary(body, target_lang)
+        body_units = [unit for unit in units if unit['kind'] == 'body']
+        translations = {}
+        token_budget = max(800, int(APP_CONFIG.get('translation_batch_tokens', 2500)))
+        batches = []
+        current = []
+        current_tokens = 0
+        for index, unit in enumerate(body_units, start=1):
+            block_id = f'{index:06d}'
+            tokens = max(1, self._count_tokens(unit['text']))
+            if current and current_tokens + tokens > token_budget:
+                batches.append(current)
+                current = []
+                current_tokens = 0
+            current.append((block_id, unit['text']))
+            current_tokens += tokens
+        if current:
+            batches.append(current)
+
+        system_prompt = self._translation_prompt(target_lang) + '\nUse this terminology glossary consistently:\n' + json.dumps(glossary, ensure_ascii=False)
+        for batch in batches:
+            self._check_cancelled()
+            marked_source = '\n\n'.join(
+                f'[[SP-P{block_id}]]\n{source}' for block_id, source in batch
+            )
+            result = self.call_deepseek(
+                system_prompt,
+                marked_source,
+                max_tokens=max(1800, min(9000, self._count_tokens(marked_source) * 2 + 1200)),
+                temp=0.0,
+                model_role='translation',
+                thinking=False,
+                allow_fallback=False,
+            )
+            result = self._remove_ai_meta_traces(result)
+            parsed = self._parse_marked_translation(result, [block_id for block_id, _ in batch])
+            for block_id, source in batch:
+                translated = parsed[block_id]
+                self._validate_translation_block(source, translated, f'SP-P{block_id}', target_lang)
+                translations[block_id] = translated
+
+        output_units = []
+        body_index = 0
+        for unit in units:
+            if unit['kind'] == 'heading':
+                translated_heading = str(glossary.get(unit['text']) or '').strip()
+                if not translated_heading:
+                    raise DeepSeekRequestError(f'Translation glossary missing heading: {unit["text"]}')
+                output_units.append('#' * int(unit['level']) + ' ' + translated_heading)
+            else:
+                body_index += 1
+                output_units.append(translations[f'{body_index:06d}'])
+
+        output = '\n\n'.join(output_units).strip()
+        if references:
+            output += f"\n\n# {self._localized_reference_heading(target_lang)}\n{references}"
+        self.validate_translation_alignment(value, output)
+        return output.strip()
+
+    @staticmethod
+    def validate_translation_alignment(english_text, translated_text):
+        source = str(english_text or '').strip()
+        target = str(translated_text or '').strip()
+        if not target:
+            raise DeepSeekRequestError('Target-language manuscript is empty.')
+        en_tree = [level for level, _ in SSciPipelineAgent._markdown_heading_tree(source)]
+        tr_tree = [level for level, _ in SSciPipelineAgent._markdown_heading_tree(target)]
+        if en_tree != tr_tree:
+            raise DeepSeekRequestError(
+                f'Translation heading-tree mismatch: English={en_tree}, target={tr_tree}.'
+            )
+        if SSciPipelineAgent._numeric_tokens(source) != SSciPipelineAgent._numeric_tokens(target):
+            raise DeepSeekRequestError('Translation numerical alignment failed.')
+        if SSciPipelineAgent._translation_citation_tokens(source) != SSciPipelineAgent._translation_citation_tokens(target):
+            raise DeepSeekRequestError('Translation citation alignment failed.')
+        if SSciPipelineAgent._translation_hypothesis_tokens(source) != SSciPipelineAgent._translation_hypothesis_tokens(target):
+            raise DeepSeekRequestError('Translation hypothesis-label alignment failed.')
+        if SSciPipelineAgent._translation_acronym_tokens(source) != SSciPipelineAgent._translation_acronym_tokens(target):
+            raise DeepSeekRequestError('Translation abbreviation alignment failed.')
+        source_body = re.split(r'(?m)^#\s+References\s*$', source, maxsplit=1)[0]
+        target_body = re.split(r'(?m)^#\s+(?:参考文献|Литература|Пайдаланылған әдебиеттер)\s*$', target, maxsplit=1)[0]
+        source_chars = len(re.sub(r'\s+', '', source_body))
+        target_chars = len(re.sub(r'\s+', '', target_body))
+        if re.search(r'(?m)^#\s+参考文献\s*$', target):
+            minimum_ratio = float(APP_CONFIG.get('translation_min_block_ratio_zh', 0.28))
+        elif re.search(r'(?m)^#\s+(?:Литература|Пайдаланылған әдебиеттер)\s*$', target):
+            minimum_ratio = float(APP_CONFIG.get('translation_min_block_ratio_cyrillic', 0.65))
+        else:
+            minimum_ratio = float(APP_CONFIG.get('translation_min_coverage_ratio', 0.25))
+        ratio = target_chars / source_chars if source_chars else 0.0
+        if source_chars and ratio < minimum_ratio:
+            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):
         """Backward-compatible strict validator; no silent citation deletion."""
         violations = self.find_multi_citation_violations(text)

core/production_guard.py

@@ -0,0 +1,182 @@
+# -*- coding: utf-8 -*-
+"""Final-output validation and packaging for SmartPaper."""
+from __future__ import annotations
+
+import hashlib
+import io
+import json
+import re
+import zipfile
+from dataclasses import dataclass
+from typing import Any
+
+from docx import Document
+
+from utils.docx_builder import DocxBuildError, text_to_docx
+
+
+class OutputValidationError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class PackageResult:
+    zip_bytes: bytes
+    file_names: tuple[str, ...]
+    manifest: str
+
+
+def replace_abstract(manuscript: str, abstract: str) -> str:
+    value = str(manuscript or "")
+    replacement = f"## Abstract\n{str(abstract or '').strip()}\n\n"
+    updated, count = re.subn(
+        r"(?ms)^##\s+Abstract\s*$\n.*?(?=^##\s+Keywords\s*$)",
+        replacement,
+        value,
+        count=1,
+    )
+    if count != 1:
+        raise OutputValidationError("English abstract block was not found exactly once.")
+    return updated
+
+
+def validate_docx_bytes(data: bytes, *, minimum_paragraphs: int = 8) -> None:
+    if not data or len(data) < 1500:
+        raise OutputValidationError("DOCX output is empty or unexpectedly small.")
+    try:
+        document = Document(io.BytesIO(data))
+    except Exception as exc:
+        raise OutputValidationError(f"DOCX cannot be reopened: {exc}") from exc
+    nonempty = [paragraph.text.strip() for paragraph in document.paragraphs if paragraph.text.strip()]
+    if len(nonempty) < minimum_paragraphs:
+        raise OutputValidationError(
+            f"DOCX contains only {len(nonempty)} nonempty paragraphs; expected at least {minimum_paragraphs}."
+        )
+
+
+def validate_markdown_manuscript(text: str, *, language_name: str) -> None:
+    value = str(text or "").strip()
+    if len(value) < 1000:
+        raise OutputValidationError(f"{language_name} manuscript is incomplete.")
+    if re.search(r"\[?CITE-\d+\]?", value, flags=re.IGNORECASE):
+        raise OutputValidationError(f"{language_name} manuscript still contains unresolved citation tags.")
+    if re.search(r"(?i)\bas an ai(?: language model)?\b|\bai[- ]generated\b", value):
+        raise OutputValidationError(f"{language_name} manuscript contains AI-writing meta-commentary.")
+    if not re.search(r"(?m)^#\s+", value):
+        raise OutputValidationError(f"{language_name} manuscript has no title/section headings.")
+
+
+def build_generation_notes(data_metadata: dict[str, Any], mode: str) -> str:
+    metadata = dict(data_metadata or {})
+    source_type = metadata.get("source_type", "unknown")
+    lines = [
+        "SmartPaper Generation Notes",
+        "",
+        f"Research mode: {mode}",
+        f"Evidence source type: {source_type}",
+        "",
+        "This file is separate from the manuscript. It documents how research evidence was supplied or generated, the applicable boundaries, and the validation steps used before packaging.",
+    ]
+    if metadata.get("disclosure_required"):
+        lines.extend(
+            [
+                "",
+                "Evidence-generation notice:",
+                "The statistical results or qualitative materials used by this run were generated by the system because verified user-provided evidence was unavailable. They were created for drafting and methodological demonstration. They must not be represented as independently collected empirical observations.",
+                "",
+                f"Generation method: {metadata.get('method') or 'Structured evidence generation under the selected quantitative or qualitative engine.'}",
+            ]
+        )
+    study_facts = metadata.get("study_facts")
+    if isinstance(study_facts, dict):
+        lines.extend(["", "Frozen qualitative study facts:", json.dumps(study_facts, ensure_ascii=False, indent=2)])
+    if metadata.get("sample_size") is not None:
+        lines.append(f"Generated sample size: {metadata.get('sample_size')}")
+    if metadata.get("evidence_count") is not None:
+        lines.append(f"Generated/retained qualitative evidence items: {metadata.get('evidence_count')}")
+    themes = metadata.get("themes")
+    if isinstance(themes, list):
+        lines.append(f"Generated/retained qualitative themes: {len(themes)}")
+    parameters = metadata.get("generation_parameters")
+    if isinstance(parameters, dict) and parameters:
+        lines.extend(
+            [
+                "",
+                "Generation parameters and process:",
+                json.dumps(parameters, ensure_ascii=False, indent=2),
+            ]
+        )
+    lines.extend(
+        [
+            "",
+            "Validation boundary:",
+            "The package confirms internal consistency, file integrity, citation resolution, and translation completeness. It does not independently verify external data collection, factual truth beyond the supplied sources, journal acceptance, or the result of any third-party AI-writing detector.",
+        ]
+    )
+    return "\n".join(lines).strip() + "\n"
+
+
+def _manifest(entries: dict[str, bytes]) -> str:
+    rows = []
+    for name in sorted(entries):
+        rows.append(f"{hashlib.sha256(entries[name]).hexdigest()}  {name}")
+    return "\n".join(rows) + "\n"
+
+
+def build_validated_package(
+    *,
+    english_markdown: str,
+    translated_markdown: str,
+    english_filename: str,
+    translated_filename: str,
+    data_metadata: dict[str, Any],
+    mode: str,
+    citation_audit: list[dict[str, Any]],
+    literature_metrics: dict[str, Any],
+) -> PackageResult:
+    validate_markdown_manuscript(english_markdown, language_name="English")
+    validate_markdown_manuscript(translated_markdown, language_name="Target-language")
+
+    try:
+        english_docx = text_to_docx(english_markdown).getvalue()
+        translated_docx = text_to_docx(translated_markdown).getvalue()
+    except DocxBuildError as exc:
+        raise OutputValidationError(str(exc)) from exc
+    validate_docx_bytes(english_docx)
+    validate_docx_bytes(translated_docx)
+
+    entries: dict[str, bytes] = {
+        english_filename: english_docx,
+        translated_filename: translated_docx,
+        "GENERATION_NOTES.txt": build_generation_notes(data_metadata, mode).encode("utf-8"),
+        "CITATION_AUDIT.json": json.dumps(citation_audit or [], ensure_ascii=False, indent=2).encode("utf-8"),
+        "LITERATURE_METRICS.json": json.dumps(literature_metrics or {}, ensure_ascii=False, indent=2).encode("utf-8"),
+    }
+    manifest = _manifest(entries)
+    entries["FILE_MANIFEST_SHA256.txt"] = manifest.encode("utf-8")
+
+    buffer = io.BytesIO()
+    with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
+        for name, content in entries.items():
+            archive.writestr(name, content)
+    zip_bytes = buffer.getvalue()
+
+    try:
+        with zipfile.ZipFile(io.BytesIO(zip_bytes), "r") as archive:
+            bad = archive.testzip()
+            if bad:
+                raise OutputValidationError(f"ZIP integrity check failed at {bad}.")
+            names = set(archive.namelist())
+            required = set(entries)
+            if names != required:
+                raise OutputValidationError(
+                    f"ZIP file list mismatch: actual={sorted(names)}, expected={sorted(required)}"
+                )
+            for name in (english_filename, translated_filename):
+                validate_docx_bytes(archive.read(name))
+    except OutputValidationError:
+        raise
+    except Exception as exc:
+        raise OutputValidationError(f"ZIP cannot be reopened: {exc}") from exc
+
+    return PackageResult(zip_bytes=zip_bytes, file_names=tuple(sorted(entries)), manifest=manifest)

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>",
@@ -222,6 +204,21 @@ Rules:
                 "analysis_method": method,
             },
             "evidence_count": len(clean_evidence),
+            "generation_parameters": {
+                "coding_approach": "Gioia-style inductive coding",
+                "requested_quote_count_when_generated": 12,
+                "maximum_retained_quotes": 15,
+                "minimum_themes": minimum_themes,
+                "minimum_quotes": minimum_quotes,
+                "consistency_constraints": [
+                    "one fixed research setting",
+                    "one fixed participant count",
+                    "one fixed interview count",
+                    "one fixed data-source set",
+                    "one fixed analysis method",
+                    "quotation tags must remain unique",
+                ],
+            },
             "evidence": clean_evidence,
             "themes": themes,
         }

core/quantitative_engine.py

@@ -192,6 +192,27 @@ class QuantitativeEngine:
                 "indicator loadings, discretization, and derived "
                 "reliability/validity summaries."
             ),
+            "generation_parameters": {
+                "constructs": [
+                    "Independent_Var",
+                    "Mediator_Var",
+                    "Dependent_Var",
+                ],
+                "sample_size": int(len(df)),
+                "likert_scale_points": 7,
+                "items_per_construct": "random integer from 3 to 5",
+                "indicator_loading_range": [0.72, 0.88],
+                "latent_correlation_control": "positive-semidefinite matrix with maximum target correlation 0.55",
+                "item_generation": "loading × latent score + normally distributed measurement error",
+                "discretization": "round to nearest integer and clamp to 1–7",
+                "reported_summaries": [
+                    "Cronbach alpha",
+                    "composite reliability",
+                    "average variance extracted",
+                    "descriptive statistics",
+                    "Fornell-Larcker matrix",
+                ],
+            },
             "evidence_sections": [
                 "Reliability and convergent validity",
                 "Descriptive statistics and discriminant validity",

desktop.ini

@@ -0,0 +1,2 @@
+[.ShellClassInfo]
+IconResource=C:\Program Files\Google\Drive File Stream\129.0.1.0\GoogleDriveFS.exe,27

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": "⏳ 系统忙碌",
@@ -339,6 +339,15 @@ TEXTS: dict[str, dict[str, str]] = {
         "account_chip_free": "免费 {value}篇",
         "account_chip_bonus": "长期 {value}篇",
         "logout_short": "退出",
+        "payment_dialog_title": "支付宝付款",
+        "payment_notice": "今日免费额度和长期额度均已用完。支付9.9元可增加3篇长期有效额度。",
+        "quota_exhausted": "今日免费额度和长期额度均已用完。请使用支付宝购买长期额度。",
+        "payment_create": "生成支付宝付款二维码",
+        "payment_creating": "正在创建支付宝订单……",
+        "payment_scan": "请使用支付宝 App 扫码支付。支付成功后点击“查询支付结果”。",
+        "payment_check": "查询支付结果",
+        "payment_paid": "支付成功，已增加3篇长期有效额度。请重新点击生成论文。",
+        "payment_expired": "付款二维码已过期，请重新生成。",
         "browse_files": "选择文件",
         "drop_docs": "将文件拖放到此处",
         "drop_data": "将文件拖放到此处",
@@ -352,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",
@@ -483,6 +492,15 @@ TEXTS: dict[str, dict[str, str]] = {
         "account_chip_free": "Free {value}",
         "account_chip_bonus": "Bonus {value}",
         "logout_short": "Sign out",
+        "payment_dialog_title": "Alipay Payment",
+        "payment_notice": "Your daily free quota and long-term quota are exhausted. Pay CNY 9.9 to add three long-term papers.",
+        "quota_exhausted": "Your daily free quota and long-term quota are exhausted. Purchase long-term quota with Alipay.",
+        "payment_create": "Create Alipay QR code",
+        "payment_creating": "Creating the Alipay order…",
+        "payment_scan": "Scan the code with the Alipay app. After payment, select “Check payment result.”",
+        "payment_check": "Check payment result",
+        "payment_paid": "Payment succeeded. Three long-term papers were added. Start the paper again.",
+        "payment_expired": "The payment QR code has expired. Create a new one.",
         "browse_files": "Choose files",
         "drop_docs": "Drag and drop files here",
         "drop_data": "Drag and drop files here",
@@ -496,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": "⏳ Система занята",
@@ -627,6 +645,15 @@ TEXTS: dict[str, dict[str, str]] = {
         "account_chip_free": "Бесплатно {value}",
         "account_chip_bonus": "Запас {value}",
         "logout_short": "Выйти",
+        "payment_dialog_title": "Оплата через Alipay",
+        "payment_notice": "Ежедневная бесплатная и долгосрочная квоты исчерпаны. Оплата 9,9 юаня добавляет квоту на три статьи без срока действия.",
+        "quota_exhausted": "Ежедневная бесплатная и долгосрочная квоты исчерпаны. Приобретите долгосрочную квоту через Alipay.",
+        "payment_create": "Создать QR-код для оплаты",
+        "payment_creating": "Создаётся заказ Alipay…",
+        "payment_scan": "Отсканируйте QR-код в приложении Alipay. После оплаты нажмите «Проверить результат оплаты».",
+        "payment_check": "Проверить результат оплаты",
+        "payment_paid": "Оплата прошла успешно. Добавлена долгосрочная квота на три статьи. Запустите создание статьи повторно.",
+        "payment_expired": "Срок действия QR-кода истёк. Создайте новый код.",
         "browse_files": "Выбрать файлы",
         "drop_docs": "Перетащите файлы сюда",
         "drop_data": "Перетащите файлы сюда",
@@ -640,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": "⏳ Жүйе бос емес",
@@ -771,6 +798,15 @@ TEXTS: dict[str, dict[str, str]] = {
         "account_chip_free": "Тегін {value}",
         "account_chip_bonus": "Қор {value}",
         "logout_short": "Шығу",
+        "payment_dialog_title": "Alipay арқылы төлеу",
+        "payment_notice": "Күнделікті тегін және ұзақ мерзімді квота таусылды. 9,9 юань төлегенде мерзімі шектелмеген үш мақала квотасы қосылады.",
+        "quota_exhausted": "Күнделікті тегін және ұзақ мерзімді квота таусылды. Alipay арқылы ұзақ мерзімді квота сатып алыңыз.",
+        "payment_create": "Төлемге арналған QR-код жасау",
+        "payment_creating": "Alipay тапсырысы жасалуда…",
+        "payment_scan": "QR-кодты Alipay қолданбасымен сканерлеңіз. Төлемнен кейін «Төлем нәтижесін тексеру» түймесін басыңыз.",
+        "payment_check": "Төлем нәтижесін тексеру",
+        "payment_paid": "Төлем сәтті өтті. Мерзімі шектелмеген үш мақала квотасы қосылды. Мақала жасауды қайта бастаңыз.",
+        "payment_expired": "QR-кодтың қолданылу мерзімі аяқталды. Жаңа код жасаңыз.",
         "browse_files": "Файлдарды таңдау",
         "drop_docs": "Файлдарды осы жерге сүйреп әкеліңіз",
         "drop_data": "Файлдарды осы жерге сүйреп әкеліңіз",

managed_auth.py

@@ -1,9 +1,10 @@
 # -*- coding: utf-8 -*-
-"""SmartPaper edge-first authentication for Streamlit.
+"""SmartPaper edge-first authentication and browser task/payment bridge.
 
-Cloudflare owns the login page and durable browser session. Streamlit receives
-one short-lived, signed assertion cookie on the proxied HTTP/WebSocket request.
-No authentication network call is made while rendering or changing language.
+Cloudflare owns the durable browser session, quota, tasks, and payments. The
+Hugging Face process verifies the signed edge assertion but does not call the
+Worker over the server-to-server network. Authenticated business operations are
+executed by the user's browser through the same-origin invisible component.
 """
 from __future__ import annotations
 
@@ -11,45 +12,23 @@ import base64
 import hashlib
 import hmac
 import html
+import io
 import json
 import os
-import socket
+import secrets
 import time
 from typing import Any, Callable
 from urllib.parse import unquote, urlsplit
 
-import requests
+import qrcode
 import streamlit as st
 
 from i18n import tr
+from smartpaper_restore_bridge import edge_bridge
 
 EDGE_ASSERTION_COOKIE_NAME = "smartpaper_edge_assertion"
 EDGE_ASSERTION_AUDIENCE = "smartpaper-streamlit"
 ASSERTION_CLOCK_SKEW_SECONDS = 30
-REQUEST_TIMEOUT = (5.0, 15.0)
-TASK_START_TIMEOUT = (8.0, 20.0)
-TASK_START_RETRY_DELAY_SECONDS = 0.5
-WORKER_DIAGNOSTIC_TIMEOUT = (5.0, 8.0)
-WORKER_DIAGNOSTIC_PREFIX = "[SP_NETWORK_DIAG]"
-
-_HTTP_SESSION = requests.Session()
-_HTTP_SESSION.headers.update({
-    "Accept": "application/json",
-    "Content-Type": "application/json",
-    "User-Agent": "SmartPaper-HuggingFace/edge-first-v1",
-})
-
-# /task/start uses a direct connection first, then one environment-aware
-# retry. Both attempts reuse the exact same idempotency_key and request body.
-_TASK_DIRECT_SESSION = requests.Session()
-_TASK_DIRECT_SESSION.trust_env = False
-_TASK_DIRECT_SESSION.headers.update({
-    "Accept": "application/json",
-    "Content-Type": "application/json",
-    "User-Agent": "SmartPaper-HuggingFace/edge-first-v1",
-})
-
-
 def _runtime_value(name: str) -> str:
     value = os.environ.get(name, "")
     if not value:
@@ -61,15 +40,21 @@ def _runtime_value(name: str) -> str:
 
 
 def _api_base() -> str:
-    # Prefer the public edge origin carried inside the verified assertion.
-    # This keeps API calls correct after a workers.dev subdomain rename and
-    # removes any dependency on a stale hard-coded public URL.
     value = str(st.session_state.get("sp_auth_api_origin") or "").strip().rstrip("/")
     if not value:
         value = _runtime_value("SMARTPAPER_AUTH_API").rstrip("/")
-    if not value.startswith("https://"):
+    parsed = urlsplit(value)
+    if (
+        parsed.scheme != "https"
+        or not parsed.netloc
+        or parsed.username is not None
+        or parsed.password is not None
+        or parsed.path not in ("", "/")
+        or parsed.query
+        or parsed.fragment
+    ):
         raise RuntimeError("SMARTPAPER_AUTH_API must be an HTTPS origin.")
-    return value
+    return f"https://{parsed.netloc}"
 
 
 def _service_key() -> str:
@@ -79,26 +64,6 @@ def _service_key() -> str:
     return value
 
 
-def _post(path: str, payload: dict[str, Any]) -> dict[str, Any]:
-    response = _HTTP_SESSION.post(
-        f"{_api_base()}{path}",
-        headers={"Authorization": f"Bearer {_service_key()}"},
-        json=payload,
-        timeout=REQUEST_TIMEOUT,
-        allow_redirects=False,
-    )
-    try:
-        data = response.json()
-    except ValueError as exc:
-        raise RuntimeError(
-            f"Authentication service returned non-JSON content (HTTP {response.status_code})."
-        ) from exc
-    if response.status_code >= 400:
-        message = data.get("message") or data.get("reason") or data.get("error") or f"HTTP {response.status_code}"
-        raise RuntimeError(str(message))
-    return data
-
-
 def _context_cookie(name: str) -> str:
     try:
         cookies = st.context.cookies
@@ -115,8 +80,7 @@ def _context_cookie(name: str) -> str:
 
 
 def _decode_b64url(value: str) -> bytes:
-    padding = "=" * (-len(value) % 4)
-    return base64.urlsafe_b64decode(value + padding)
+    return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
 
 
 def _verify_edge_assertion(token: str) -> dict[str, Any]:
@@ -139,45 +103,46 @@ def _verify_edge_assertion(token: str) -> dict[str, Any]:
     if payload.get("ver") != 1 or payload.get("aud") != EDGE_ASSERTION_AUDIENCE:
         raise RuntimeError("invalid_edge_assertion_claims")
     api_origin = str(payload.get("api_origin") or "").strip().rstrip("/")
-    parsed_origin = urlsplit(api_origin)
+    parsed = urlsplit(api_origin)
     if (
-        parsed_origin.scheme != "https"
-        or not parsed_origin.netloc
-        or parsed_origin.username is not None
-        or parsed_origin.password is not None
-        or parsed_origin.path not in ("", "/")
-        or parsed_origin.query
-        or parsed_origin.fragment
+        parsed.scheme != "https"
+        or not parsed.netloc
+        or parsed.username is not None
+        or parsed.password is not None
+        or parsed.path not in ("", "/")
+        or parsed.query
+        or parsed.fragment
     ):
         raise RuntimeError("invalid_edge_assertion_origin")
-    api_origin = f"https://{parsed_origin.netloc}"
     issued_at = int(payload.get("iat") or 0)
     expires_at = int(payload.get("exp") or 0)
     if issued_at > now + ASSERTION_CLOCK_SKEW_SECONDS or expires_at <= now - ASSERTION_CLOCK_SKEW_SECONDS:
         raise RuntimeError("expired_edge_assertion")
     if expires_at - issued_at > 9 * 60 * 60:
         raise RuntimeError("invalid_edge_assertion_lifetime")
-    browser_token = str(payload.get("browser_token") or "").strip()
     user = payload.get("user") or {}
-    if not browser_token or not isinstance(user, dict) or not user.get("id") or user.get("status") != "active":
+    if not isinstance(user, dict) or not user.get("id") or user.get("status") != "active":
         raise RuntimeError("invalid_edge_assertion_identity")
-    payload["api_origin"] = api_origin
+    payload["api_origin"] = f"https://{parsed.netloc}"
     return payload
 
 
 def _clear_generation_session_state() -> None:
     for key in (
-        "sp_generation_task", "sp_current_node", "sp_last_task_heartbeat",
-        "zip_buffer", "zip_filename", "is_running", "start_pipeline",
-        "resume_pipeline", "run_mode", "rag_engine", "up_amj_ref",
-        "up_amj_data", "up_asq_ref", "up_asq_data",
+        "sp_generation_task", "sp_tracked_task_id", "sp_current_node",
+        "sp_task_recovery_initialized", "sp_last_ui_task_status",
+        "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",
+        "up_asq_data", "sp_payment_required", "sp_payment_order",
+        "sp_payment_request_id",
     ):
         st.session_state.pop(key, None)
 
 
 def _clear_session() -> None:
     for key in (
-        "sp_session_token", "sp_session_expires_at", "sp_user",
+        "sp_session_expires_at", "sp_user",
         "sp_last_verified_at", "sp_edge_assertion", "sp_auth_api_origin",
     ):
         st.session_state.pop(key, None)
@@ -198,7 +163,6 @@ def require_alipay_login(language: str = "zh") -> dict[str, Any]:
         st.stop()
     user = dict(payload["user"])
     st.session_state["sp_edge_assertion"] = token
-    st.session_state["sp_session_token"] = payload["browser_token"]
     st.session_state["sp_session_expires_at"] = payload["exp"]
     st.session_state["sp_user"] = user
     st.session_state["sp_auth_api_origin"] = str(payload["api_origin"]).rstrip("/")
@@ -206,237 +170,175 @@ def require_alipay_login(language: str = "zh") -> dict[str, Any]:
     return user
 
 
-def _authorized_post(path: str, payload: dict[str, Any] | None = None) -> dict[str, Any]:
-    session_token = str(st.session_state.get("sp_session_token") or "").strip()
-    if not session_token:
-        raise RuntimeError("session_token_required")
-    body = dict(payload or {})
-    body["session_token"] = session_token
-    return _post(path, body)
-
-
-def _post_task_start(payload: dict[str, Any]) -> dict[str, Any]:
-    """Start one task with one bounded retry and the same idempotency key."""
-    url = f"{_api_base()}/task/start"
-    headers = {"Authorization": f"Bearer {_service_key()}"}
-    transports = (_TASK_DIRECT_SESSION, _HTTP_SESSION)
-    last_error: Exception | None = None
-
-    for attempt, session in enumerate(transports):
-        try:
-            response = session.post(
-                url,
-                headers=headers,
-                json=payload,
-                timeout=TASK_START_TIMEOUT,
-                allow_redirects=False,
-            )
-        except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as exc:
-            last_error = exc
-            if attempt == 0:
-                time.sleep(TASK_START_RETRY_DELAY_SECONDS)
-                continue
-            raise
-
-        try:
-            data = response.json()
-        except ValueError as exc:
-            raise RuntimeError(
-                f"Authentication service returned non-JSON content (HTTP {response.status_code})."
-            ) from exc
-
-        if response.status_code >= 400:
-            message = (
-                data.get("message")
-                or data.get("reason")
-                or data.get("error")
-                or f"HTTP {response.status_code}"
-            )
-            raise RuntimeError(str(message))
-
-        return data
-
-    if last_error is not None:
-        raise last_error
-    raise RuntimeError("task_start_request_failed")
-
-
-def _diagnostic_log(event: str, **fields: Any) -> None:
-    record = {
-        "event": event,
-        "time_unix": round(time.time(), 3),
-        **fields,
-    }
-    print(
-        f"{WORKER_DIAGNOSTIC_PREFIX} "
-        + json.dumps(record, ensure_ascii=False, sort_keys=True, default=str),
-        flush=True,
+def _bridge_key(prefix: str, value: str) -> str:
+    digest = hashlib.sha256(str(value).encode("utf-8")).hexdigest()[:20]
+    return f"{prefix}_{digest}"
+
+
+def _edge_call(
+    action: str,
+    command_id: str,
+    payload: dict[str, Any] | None = None,
+    *,
+    key_prefix: str = "sp_edge",
+) -> dict[str, Any] | None:
+    return edge_bridge(
+        action=action,
+        command_id=command_id,
+        api_origin=_api_base(),
+        payload=payload or {},
+        key=_bridge_key(key_prefix, command_id),
     )
 
 
-def _run_worker_health_diagnostic() -> dict[str, Any]:
-    """Test Hugging Face -> Worker connectivity without creating a task."""
-    base_url = _api_base()
-    parsed = urlsplit(base_url)
-    host = str(parsed.hostname or "").strip()
-    port = parsed.port or 443
-    health_url = f"{base_url}/health"
-
-    summary: dict[str, Any] = {
-        "base_url": base_url,
-        "dns_ok": False,
-        "direct_ok": False,
-        "environment_ok": False,
-    }
-
-    dns_started = time.perf_counter()
-    try:
-        answers = socket.getaddrinfo(
-            host,
-            port,
-            family=socket.AF_UNSPEC,
-            type=socket.SOCK_STREAM,
-        )
-        addresses = sorted({
-            str(item[4][0])
-            for item in answers
-            if item and len(item) >= 5 and item[4]
-        })
-        elapsed_ms = round((time.perf_counter() - dns_started) * 1000, 1)
-        summary["dns_ok"] = bool(addresses)
-        summary["dns_addresses"] = addresses[:8]
-        _diagnostic_log(
-            "dns_result",
-            host=host,
-            port=port,
-            ok=bool(addresses),
-            elapsed_ms=elapsed_ms,
-            addresses=addresses[:8],
-        )
-    except Exception as exc:
-        elapsed_ms = round((time.perf_counter() - dns_started) * 1000, 1)
-        summary["dns_error"] = type(exc).__name__
-        _diagnostic_log(
-            "dns_error",
-            host=host,
-            port=port,
-            elapsed_ms=elapsed_ms,
-            exception_type=type(exc).__name__,
-            message=str(exc)[:500],
-        )
-
-    transports = (
-        ("direct", _TASK_DIRECT_SESSION),
-        ("environment", _HTTP_SESSION),
+def _raise_edge_error(result: dict[str, Any], fallback: str) -> None:
+    message = (
+        result.get("message")
+        or result.get("reason")
+        or result.get("error")
+        or fallback
     )
+    raise RuntimeError(str(message))
 
-    for transport_name, session in transports:
-        started = time.perf_counter()
-        _diagnostic_log(
-            "health_request_start",
-            transport=transport_name,
-            url=health_url,
-            timeout_connect=WORKER_DIAGNOSTIC_TIMEOUT[0],
-            timeout_read=WORKER_DIAGNOSTIC_TIMEOUT[1],
-        )
-        try:
-            response = session.get(
-                health_url,
-                timeout=WORKER_DIAGNOSTIC_TIMEOUT,
-                allow_redirects=False,
-                headers={
-                    "Accept": "application/json",
-                    "User-Agent": "SmartPaper-HuggingFace/network-diagnostic-r2",
-                },
-            )
-            elapsed_ms = round((time.perf_counter() - started) * 1000, 1)
-            ok = 200 <= response.status_code < 300
-            summary[f"{transport_name}_ok"] = ok
-            summary[f"{transport_name}_status"] = response.status_code
-            _diagnostic_log(
-                "health_response",
-                transport=transport_name,
-                ok=ok,
-                status_code=response.status_code,
-                elapsed_ms=elapsed_ms,
-                content_type=str(response.headers.get("content-type") or ""),
-                cf_ray=str(response.headers.get("cf-ray") or ""),
-                server=str(response.headers.get("server") or ""),
-                body_prefix=str(response.text or "")[:300],
-            )
-        except requests.exceptions.RequestException as exc:
-            elapsed_ms = round((time.perf_counter() - started) * 1000, 1)
-            summary[f"{transport_name}_error"] = type(exc).__name__
-            _diagnostic_log(
-                "health_error",
-                transport=transport_name,
-                elapsed_ms=elapsed_ms,
-                exception_type=type(exc).__name__,
-                message=str(exc)[:500],
-            )
-        except Exception as exc:
-            elapsed_ms = round((time.perf_counter() - started) * 1000, 1)
-            summary[f"{transport_name}_error"] = type(exc).__name__
-            _diagnostic_log(
-                "health_unexpected_error",
-                transport=transport_name,
-                elapsed_ms=elapsed_ms,
-                exception_type=type(exc).__name__,
-                message=str(exc)[:500],
-            )
-
-    _diagnostic_log("diagnostic_summary", **summary)
-    return summary
-
-
-def start_generation_task(*, idempotency_key: str, mode: str, interface_language: str) -> dict[str, Any]:
-    """Temporary R2 diagnostic mode: test /health only; never call /task/start."""
-    session_token = str(st.session_state.get("sp_session_token") or "").strip()
-    if not session_token:
-        raise RuntimeError("session_token_required")
-
-    summary = _run_worker_health_diagnostic()
-    if summary.get("direct_ok") or summary.get("environment_ok"):
-        raise RuntimeError(
-            "R2网络诊断完成：Worker /health 可访问。请查看 Hugging Face Runtime Logs 中的 "
-            "[SP_NETWORK_DIAG] 记录。当前未创建论文任务，也未预留额度。"
-        )
-    raise RuntimeError(
-        "R2网络诊断完成：Worker /health 不可访问。请查看 Hugging Face Runtime Logs 中的 "
-        "[SP_NETWORK_DIAG] 记录。当前未创建论文任务，也未预留额度。"
+
+def _merge_quota_into_session(result: dict[str, Any]) -> None:
+    """Refresh the approved account card immediately after a quota mutation."""
+    quota = result.get("quota")
+    if not isinstance(quota, dict):
+        return
+    user = dict(st.session_state.get("sp_user") or {})
+    for key in (
+        "quota_date",
+        "daily_free_limit",
+        "daily_free_reserved",
+        "daily_free_used",
+        "daily_free_remaining",
+        "bonus_papers",
+    ):
+        if key in quota:
+            user[key] = quota[key]
+    if user:
+        st.session_state["sp_user"] = user
+        st.session_state["sp_account_revision"] = int(
+            st.session_state.get("sp_account_revision") or 0
+        ) + 1
+
+
+def get_account_snapshot(*, block: bool = False) -> dict[str, Any] | None:
+    bucket = int(time.time() // 30)
+    revision = int(st.session_state.get("sp_account_revision") or 0)
+    command_id = f"account:{revision}:{bucket}"
+    result = _edge_call("account", command_id, {}, key_prefix="sp_account")
+    if result is None:
+        if block:
+            st.stop()
+        return None
+    if not result.get("ok"):
+        if block:
+            _raise_edge_error(result, "account_query_failed")
+        return None
+    user = result.get("user")
+    if isinstance(user, dict):
+        st.session_state["sp_user"] = dict(user)
+    # R3.1 single-writer rule: account refresh owns identity/quota only.
+    # Task UI state is initialized by recover_generation_task() and thereafter
+    # updated only by generation_task_status()/explicit task actions.
+    return result
+
+
+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")
+    result = _edge_call(
+        "task_start",
+        normalized,
+        {
+            "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",
     )
+    if result is None:
+        st.stop()
+    if not result.get("ok"):
+        _raise_edge_error(result, "task_start_failed")
+    _merge_quota_into_session(result)
+    started_task = result.get("task")
+    if isinstance(started_task, dict) and started_task.get("task_id"):
+        st.session_state["sp_tracked_task_id"] = str(started_task["task_id"])
+    return dict(result)
 
 
 def recover_generation_task() -> dict[str, Any]:
-    return _authorized_post("/task/recover", {})
+    """Initialize current task state once from the authoritative Worker.
 
-
-def heartbeat_generation_task(task_id: str, *, current_node: str = "") -> dict[str, Any]:
-    return _authorized_post("/task/heartbeat", {"task_id": task_id, "current_node": current_node})
+    Account polling is intentionally forbidden from writing task state. This
+    recovery call is the only fresh-session initializer; later changes come
+    from generation_task_status() or explicit start/cancel operations.
+    """
+    command_id = f"recover:{int(time.time() // 10)}"
+    result = _edge_call("task_recover", command_id, {}, key_prefix="sp_task_recover")
+    if result is None:
+        st.stop()
+    if not result.get("ok"):
+        _raise_edge_error(result, "task_recover_failed")
+    _merge_quota_into_session(result)
+    task = result.get("task")
+    if isinstance(task, dict) and task.get("task_id"):
+        st.session_state["sp_generation_task"] = dict(task)
+        st.session_state["sp_tracked_task_id"] = str(task["task_id"])
+        st.session_state["sp_last_ui_task_status"] = str(task.get("status") or "")
+    else:
+        st.session_state["sp_generation_task"] = None
+        st.session_state["sp_last_ui_task_status"] = ""
+        st.session_state.pop("sp_tracked_task_id", None)
+    return dict(result)
 
 
 def generation_task_status(task_id: str) -> dict[str, Any]:
-    return _authorized_post("/task/status", {"task_id": task_id})
-
-
-def finish_generation_task(task_id: str, *, status: str, safe_error_code: str = "", safe_error_detail: str = "", current_node: str = "") -> dict[str, Any]:
-    return _authorized_post("/task/finish", {"task_id": task_id, "status": status, "safe_error_code": safe_error_code, "safe_error_detail": safe_error_detail, "current_node": current_node})
+    normalized = str(task_id or "").strip()
+    command_id = f"status:{normalized}:{int(time.time() // 5)}"
+    result = _edge_call(
+        "task_status", command_id, {"task_id": normalized}, key_prefix="sp_task_status"
+    )
+    if result is None:
+        st.stop()
+    if not result.get("ok"):
+        _raise_edge_error(result, "task_status_failed")
+    _merge_quota_into_session(result)
+    task = result.get("task")
+    if isinstance(task, dict):
+        st.session_state["sp_generation_task"] = dict(task)
+        if task.get("task_id"):
+            st.session_state["sp_tracked_task_id"] = str(task["task_id"])
+    return dict(result)
 
 
 def cancel_generation_task(task_id: str | None = None) -> dict[str, Any]:
-    payload: dict[str, Any] = {}
-    if task_id:
-        payload["task_id"] = task_id
-    return _authorized_post("/task/cancel", payload)
+    """Cancel synchronously so quota release is confirmed before local cleanup."""
+    normalized = str(task_id or "").strip()
+    command_id = f"cancel:{normalized or 'latest'}"
+    result = _edge_call(
+        "task_cancel",
+        command_id,
+        {"task_id": normalized} if normalized else {},
+        key_prefix="sp_task_cancel",
+    )
+    if result is None:
+        st.stop()
+    if not result.get("ok"):
+        _raise_edge_error(result, "task_cancel_failed")
+    reservation = result.get("reservation")
+    if result.get("cancelled") and reservation and str(reservation.get("status") or "").lower() != "released":
+        raise RuntimeError("task_cancel_quota_not_released")
+    _merge_quota_into_session(result)
+    return dict(result)
 
 
 def logout_alipay_session(on_logout: Callable[[str, str], None] | None = None) -> None:
-    """Clear only the current Streamlit state.
-
-    Durable logout is performed by the browser navigating directly to the
-    Cloudflare edge endpoint ``/edge/logout``. This helper remains for API
-    compatibility and must not be used as the visible logout control.
-    """
     user_id = str((st.session_state.get("sp_user") or {}).get("id") or "")
     task = st.session_state.get("sp_generation_task") or {}
     task_id = str(task.get("task_id") or "")
@@ -448,15 +350,115 @@ def logout_alipay_session(on_logout: Callable[[str, str], None] | None = None) -
     _clear_session()
     _clear_generation_session_state()
 
+
+def create_payment_order(idempotency_key: str) -> dict[str, Any]:
+    normalized = str(idempotency_key or "").strip()
+    result = _edge_call(
+        "pay_create",
+        normalized,
+        {"idempotency_key": normalized},
+        key_prefix="sp_pay_create",
+    )
+    if result is None:
+        st.stop()
+    if not result.get("ok"):
+        _raise_edge_error(result, "payment_create_failed")
+    return dict(result)
+
+
+def payment_order_status(merchant_order_no: str) -> dict[str, Any]:
+    order_no = str(merchant_order_no or "").strip()
+    command_id = f"pay-status:{order_no}:{int(time.time() // 3)}"
+    result = _edge_call(
+        "pay_status",
+        command_id,
+        {"merchant_order_no": order_no},
+        key_prefix="sp_pay_status",
+    )
+    if result is None:
+        st.stop()
+    if not result.get("ok"):
+        _raise_edge_error(result, "payment_status_failed")
+    return dict(result)
+
+
+def _qr_png_bytes(value: str) -> bytes:
+    image = qrcode.make(str(value))
+    buffer = io.BytesIO()
+    image.save(buffer, format="PNG")
+    return buffer.getvalue()
+
+
+def _render_payment_purchase_body(language: str) -> None:
+    st.warning(tr(language, "payment_notice"))
+
+    if st.button(tr(language, "payment_create"), key="sp_create_payment_order"):
+        st.session_state["sp_payment_request_id"] = (
+            f"pay_{int(time.time() * 1000)}_{secrets.token_urlsafe(12)}"
+        )
+
+    request_id = str(st.session_state.get("sp_payment_request_id") or "").strip()
+    order = st.session_state.get("sp_payment_order")
+    if request_id and not order:
+        st.info(tr(language, "payment_creating"))
+        order_result = create_payment_order(request_id)
+        order = dict(order_result.get("order") or {})
+        if not order.get("merchant_order_no") or not order.get("qr_code"):
+            raise RuntimeError("payment_order_missing_qr_code")
+        st.session_state["sp_payment_order"] = order
+
+    order = st.session_state.get("sp_payment_order") or {}
+    qr_code = str(order.get("qr_code") or "").strip()
+    if qr_code:
+        st.image(_qr_png_bytes(qr_code), width=280)
+        st.caption(tr(language, "payment_scan"))
+        if st.button(tr(language, "payment_check"), key="sp_check_payment_status"):
+            status_result = payment_order_status(order.get("merchant_order_no", ""))
+            updated = dict(status_result.get("order") or {})
+            st.session_state["sp_payment_order"] = updated
+            if updated.get("status") == "paid" and updated.get("credit_applied"):
+                user = status_result.get("user")
+                if isinstance(user, dict):
+                    st.session_state["sp_user"] = dict(user)
+                    st.session_state["sp_account_revision"] = int(
+                        st.session_state.get("sp_account_revision") or 0
+                    ) + 1
+                st.session_state["sp_payment_required"] = False
+                st.session_state["sp_payment_order"] = None
+                st.session_state["sp_payment_request_id"] = ""
+                st.success(tr(language, "payment_paid"))
+            elif updated.get("status") in {"closed", "failed"}:
+                st.session_state["sp_payment_order"] = None
+                st.session_state["sp_payment_request_id"] = ""
+                st.warning(tr(language, "payment_expired"))
+
+
+def render_payment_purchase(language: str) -> None:
+    """Show payment in a modal so the approved normal-page layout is unchanged."""
+    dialog_factory = getattr(st, "dialog", None)
+    if callable(dialog_factory):
+        @dialog_factory(tr(language, "payment_dialog_title"))
+        def _payment_dialog():
+            _render_payment_purchase_body(language)
+        _payment_dialog()
+        return
+    # Compatibility fallback for older Streamlit builds.
+    _render_payment_purchase_body(language)
+
+
 def render_logged_in_account(
     language: str,
     user: dict[str, Any] | None = None,
     on_logout: Callable[[str, str], None] | None = None,
 ) -> None:
-    """Render a responsive account summary in the top toolbar."""
-    user = user or st.session_state.get("sp_user", {})
+    """Render the approved responsive account summary with live quota values."""
+    live = get_account_snapshot(block=False)
+    if live and isinstance(live.get("user"), dict):
+        user = dict(live["user"])
+    else:
+        user = user or st.session_state.get("sp_user", {})
     user_id = user.get("id", "—")
-    free_limit = user.get("daily_free_limit", 3)
+    free_remaining = user.get("daily_free_remaining", user.get("daily_free_limit", 3))
     bonus_papers = user.get("bonus_papers", 0)
 
     st.markdown(
@@ -578,21 +580,19 @@ def render_logged_in_account(
     )
 
     info_col, logout_col = st.columns([5.9, 1.65])
-
     with info_col:
         st.markdown(
             f"""
             <div class="sp-account-card">
-              <span>{html.escape(tr(language, "account_chip_user", value=user_id))}</span>
+              <span>{html.escape(tr(language, 'account_chip_user', value=user_id))}</span>
               <span class="sp-account-separator">•</span>
-              <span>{html.escape(tr(language, "account_chip_free", value=free_limit))}</span>
+              <span>{html.escape(tr(language, 'account_chip_free', value=free_remaining))}</span>
               <span class="sp-account-separator">•</span>
-              <span>{html.escape(tr(language, "account_chip_bonus", value=bonus_papers))}</span>
+              <span>{html.escape(tr(language, 'account_chip_bonus', value=bonus_papers))}</span>
             </div>
             """,
             unsafe_allow_html=True,
         )
-
     with logout_col:
         logout_label = html.escape(tr(language, "logout_short"))
         logout_help = html.escape(tr(language, "logout"), quote=True)

requirements.txt

@@ -1,17 +1,14 @@
 streamlit==1.38.0
-requests
-python-docx
-pandas
-numpy
-statsmodels
-tiktoken
-tenacity
-langdetect
-mistune>=3.0.0
-beautifulsoup4
-PyMuPDF
-openpyxl
-tabulate
-pycryptodome
-qrcode[pil]>=8.0,<9
-streamlit-autorefresh==1.0.1
+requests>=2.31,<3
+python-docx>=1.1,<2
+pandas>=2.1,<3
+numpy>=1.26,<3
+scikit-learn>=1.4,<2
+tiktoken>=0.7,<1
+langdetect==1.0.9
+mistune>=3,<4
+beautifulsoup4>=4.12,<5
+PyMuPDF>=1.24,<2
+openpyxl>=3.1,<4
+tabulate>=0.9,<1
+qrcode[pil]>=8,<9

smartpaper_restore_bridge/__init__.py

@@ -1,8 +1,10 @@
 # -*- coding: utf-8 -*-
-"""Invisible browser storage bridge used by SmartPaper authentication.
+"""Invisible browser bridge for SmartPaper recovery and edge API calls.
 
-The component is intentionally dependency-free. Its frontend is served by
-Streamlit itself and returns only JSON-serializable status data to Python.
+The component is served by Streamlit through the public Cloudflare origin. It
+therefore allows the authenticated browser to call same-origin Worker endpoints
+with the existing HttpOnly browser cookie. No service key or cookie value is
+passed to Python or JavaScript arguments.
 """
 
 from __future__ import annotations
@@ -31,18 +33,42 @@ def restore_bridge(
     expires_at: str = "",
     key: str = "sp_restore_bridge",
 ) -> dict[str, Any] | None:
-    """Read, write, or delete SmartPaper's browser recovery token.
-
-    The return value is asynchronous: the first call normally returns ``None``;
-    after the frontend completes the requested storage operation, Streamlit
-    reruns the script and this function returns a result dictionary.
-    """
-
+    """Backward-compatible recovery-token storage operation."""
     value = _COMPONENT(
         action=str(action or "read"),
         command_id=str(command_id or ""),
         token=str(token or ""),
         expires_at=str(expires_at or ""),
+        api_origin="",
+        payload={},
+        height=0,
+        default=None,
+        key=key,
+    )
+    return value if isinstance(value, dict) else None
+
+
+def edge_bridge(
+    *,
+    action: str,
+    command_id: str,
+    api_origin: str,
+    payload: dict[str, Any] | None = None,
+    key: str = "sp_edge_bridge",
+) -> dict[str, Any] | None:
+    """Execute one authenticated same-origin Worker command in the browser.
+
+    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 ""),
+        command_id=str(command_id or ""),
+        token="",
+        expires_at="",
+        api_origin=str(api_origin or ""),
+        payload=dict(payload or {}),
         height=0,
         default=None,
         key=key,

smartpaper_restore_bridge/desktop.ini

@@ -0,0 +1,2 @@
+[.ShellClassInfo]
+IconResource=C:\Program Files\Google\Drive File Stream\129.0.1.0\GoogleDriveFS.exe,27

smartpaper_restore_bridge/frontend/desktop.ini

@@ -0,0 +1,2 @@
+[.ShellClassInfo]
+IconResource=C:\Program Files\Google\Drive File Stream\129.0.1.0\GoogleDriveFS.exe,27

smartpaper_restore_bridge/frontend/index.html

@@ -3,18 +3,9 @@
 <head>
   <meta charset="utf-8" />
   <meta name="viewport" content="width=device-width, initial-scale=1" />
-  <title>SmartPaper Restore Bridge</title>
+  <title>SmartPaper Browser Bridge</title>
   <style>
-    html, body {
-      width: 0;
-      height: 0;
-      min-width: 0;
-      min-height: 0;
-      margin: 0;
-      padding: 0;
-      overflow: hidden;
-      background: transparent;
-    }
+    html, body { width:0; height:0; min-width:0; min-height:0; margin:0; padding:0; overflow:hidden; background:transparent; }
   </style>
 </head>
 <body>
@@ -22,44 +13,56 @@
 (() => {
   "use strict";
 
-  const STORAGE_KEY = "smartpaper.restore_token.v1";
+  const RESTORE_KEY = "smartpaper.restore_token.v1";
   const LEGACY_COOKIE_NAME = "__Host-smartpaper_restore";
+  const RESULT_PREFIX = "smartpaper.edge.result.v3:";
   const MAX_TOKEN_LENGTH = 4096;
+  const MAX_RESULT_AGE_MS = 10 * 60 * 1000;
+  const REQUEST_TIMEOUT_MS = 20000;
   let lastResponseSignature = "";
 
+  const ACTION_PATHS = Object.freeze({
+    task_start: "/edge/task/start",
+    task_cancel: "/edge/task/cancel",
+    task_status: "/edge/task/status",
+    task_recover: "/edge/task/recover",
+    account: "/edge/account",
+    pay_create: "/edge/pay/create",
+    pay_status: "/edge/pay/status"
+  });
+
   function sendMessage(type, data) {
-    const message = Object.assign(
-      { isStreamlitMessage: true, type: type },
-      data || {}
-    );
-    window.parent.postMessage(message, "*");
+    window.parent.postMessage(Object.assign({ isStreamlitMessage:true, type }, data || {}), "*");
   }
 
-  function setFrameHeight(height) {
-    sendMessage("streamlit:setFrameHeight", { height: height });
-  }
+  function setFrameHeight() { sendMessage("streamlit:setFrameHeight", { height:0 }); }
 
   function sendValue(value) {
     const signature = JSON.stringify(value);
-    if (signature === lastResponseSignature) {
-      return;
-    }
+    if (signature === lastResponseSignature) return;
     lastResponseSignature = signature;
-    sendMessage("streamlit:setComponentValue", { value: value });
+    sendMessage("streamlit:setComponentValue", { value });
+  }
+
+  function readJson(key, fallback=null) {
+    try {
+      const raw = window.localStorage.getItem(key);
+      return raw ? JSON.parse(raw) : fallback;
+    } catch (_) { return fallback; }
+  }
+
+  function writeJson(key, value) {
+    try { window.localStorage.setItem(key, JSON.stringify(value)); return true; }
+    catch (_) { return false; }
   }
 
   function normalizedToken(value) {
     const token = typeof value === "string" ? value.trim() : "";
-    if (!token || token.length > MAX_TOKEN_LENGTH) {
-      return "";
-    }
-    return token;
+    return token && token.length <= MAX_TOKEN_LENGTH ? token : "";
   }
 
   function parseExpiry(value) {
-    if (typeof value !== "string" || !value.trim()) {
-      return null;
-    }
+    if (typeof value !== "string" || !value.trim()) return null;
     const timestamp = Date.parse(value);
     return Number.isFinite(timestamp) ? timestamp : null;
   }
@@ -67,178 +70,147 @@
   function readLegacyCookie() {
     try {
       const prefix = `${LEGACY_COOKIE_NAME}=`;
-      const entries = document.cookie ? document.cookie.split("; ") : [];
-      const entry = entries.find((item) => item.startsWith(prefix));
-      if (!entry) {
-        return "";
-      }
-      return normalizedToken(decodeURIComponent(entry.slice(prefix.length)));
-    } catch (_) {
-      return "";
-    }
+      const entry = (document.cookie ? document.cookie.split("; ") : []).find(item => item.startsWith(prefix));
+      return entry ? normalizedToken(decodeURIComponent(entry.slice(prefix.length))) : "";
+    } catch (_) { return ""; }
   }
 
   function deleteLegacyCookie() {
-    try {
-      document.cookie = `${LEGACY_COOKIE_NAME}=; Max-Age=0; Path=/; Secure; SameSite=Strict`;
-    } catch (_) {
-      // The legacy cookie is optional. Storage operations continue normally.
-    }
+    try { document.cookie = `${LEGACY_COOKIE_NAME}=; Max-Age=0; Path=/; Secure; SameSite=Strict`; }
+    catch (_) {}
   }
 
   function readStoredRecord() {
-    const raw = window.localStorage.getItem(STORAGE_KEY);
-    if (!raw) {
-      return null;
-    }
-
-    let record;
-    try {
-      record = JSON.parse(raw);
-    } catch (_) {
-      window.localStorage.removeItem(STORAGE_KEY);
-      return null;
-    }
-
+    const record = readJson(RESTORE_KEY, null);
     const token = normalizedToken(record && record.token);
-    if (!token) {
-      window.localStorage.removeItem(STORAGE_KEY);
-      return null;
-    }
-
-    const expiresAt =
-      record && typeof record.expires_at === "string"
-        ? record.expires_at
-        : "";
-    const expiryTimestamp = parseExpiry(expiresAt);
-    if (expiresAt.trim() && expiryTimestamp === null) {
-      window.localStorage.removeItem(STORAGE_KEY);
-      return null;
-    }
-    if (expiryTimestamp !== null && expiryTimestamp <= Date.now()) {
-      window.localStorage.removeItem(STORAGE_KEY);
-      return null;
+    if (!token) { window.localStorage.removeItem(RESTORE_KEY); return null; }
+    const expiresAt = record && typeof record.expires_at === "string" ? record.expires_at : "";
+    const expiry = parseExpiry(expiresAt);
+    if ((expiresAt.trim() && expiry === null) || (expiry !== null && expiry <= Date.now())) {
+      window.localStorage.removeItem(RESTORE_KEY); return null;
     }
-
-    return { token: token, expires_at: expiresAt };
+    return { token, expires_at:expiresAt };
   }
 
   function writeStoredRecord(token, expiresAt) {
     const normalized = normalizedToken(token);
-    if (!normalized) {
-      throw new Error("invalid_token");
-    }
-
-    const record = {
-      token: normalized,
-      expires_at: typeof expiresAt === "string" ? expiresAt : "",
-      stored_at: new Date().toISOString(),
-    };
-    window.localStorage.setItem(STORAGE_KEY, JSON.stringify(record));
-
+    if (!normalized) throw new Error("invalid_token");
+    const record = { token:normalized, expires_at:typeof expiresAt === "string" ? expiresAt : "", stored_at:new Date().toISOString() };
+    if (!writeJson(RESTORE_KEY, record)) throw new Error("storage_unavailable");
     const verified = readStoredRecord();
-    if (!verified || verified.token !== normalized) {
-      throw new Error("write_verification_failed");
-    }
+    if (!verified || verified.token !== normalized) throw new Error("write_verification_failed");
     deleteLegacyCookie();
     return verified;
   }
 
-  function handleRead(commandId) {
-    let record = readStoredRecord();
-    let source = "local_storage";
-
+  function handleRestore(action, commandId, args) {
+    if (action === "write") {
+      const record = writeStoredRecord(args.token || "", args.expires_at || "");
+      return { action, command_id:commandId, status:"written", expires_at:record.expires_at || "" };
+    }
+    if (action === "delete") {
+      window.localStorage.removeItem(RESTORE_KEY); deleteLegacyCookie();
+      const deleted = window.localStorage.getItem(RESTORE_KEY) === null;
+      return { action, command_id:commandId, status:deleted ? "deleted" : "error", error:deleted ? "" : "delete_verification_failed" };
+    }
+    let record = readStoredRecord(); let source = "local_storage";
     if (!record) {
-      const legacyToken = readLegacyCookie();
-      if (legacyToken) {
-        record = writeStoredRecord(legacyToken, "");
-        source = "legacy_cookie";
-      }
+      const legacy = readLegacyCookie();
+      if (legacy) { record = writeStoredRecord(legacy, ""); source = "legacy_cookie"; }
     }
+    return record
+      ? { action:"read", command_id:commandId, status:"found", token:record.token, expires_at:record.expires_at || "", source }
+      : { action:"read", command_id:commandId, status:"missing" };
+  }
 
-    if (!record) {
+  function normalizeOrigin(value) {
+    try {
+      const parsed = new URL(String(value || ""));
+      if (parsed.protocol !== "https:") return "";
+      const parentOrigin = window.parent.location.origin;
+      return parsed.origin === parentOrigin ? parsed.origin : "";
+    } catch (_) { return ""; }
+  }
+
+  async function postJson(apiOrigin, path, payload) {
+    const origin = normalizeOrigin(apiOrigin);
+    if (!origin) return { ok:false, error:"invalid_api_origin", http_status:0 };
+    const controller = new AbortController();
+    const timer = window.setTimeout(() => controller.abort(), REQUEST_TIMEOUT_MS);
+    try {
+      const response = await fetch(new URL(path, origin).toString(), {
+        method:"POST",
+        credentials:"include",
+        cache:"no-store",
+        redirect:"error",
+        headers:{ "Accept":"application/json", "Content-Type":"application/json", "X-SmartPaper-Client":"browser-bridge-v1" },
+        body:JSON.stringify(payload || {}),
+        signal:controller.signal
+      });
+      const text = await response.text();
+      let data = {};
+      try { data = text ? JSON.parse(text) : {}; }
+      catch (_) { data = { ok:false, error:"non_json_response", body_prefix:text.slice(0,200) }; }
+      return Object.assign({}, data, { ok:response.ok && data.ok !== false, http_status:response.status });
+    } catch (error) {
       return {
-        action: "read",
-        command_id: commandId,
-        status: "missing",
+        ok:false,
+        error:error && error.name === "AbortError" ? "browser_edge_timeout" : "browser_edge_request_failed",
+        detail:String(error || "").slice(0,300),
+        http_status:0
       };
-    }
+    } finally { window.clearTimeout(timer); }
+  }
 
-    return {
-      action: "read",
-      command_id: commandId,
-      status: "found",
-      token: record.token,
-      expires_at: record.expires_at || "",
-      source: source,
-    };
+  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;
+    return cached.value || null;
   }
 
-  function handleWrite(commandId, token, expiresAt) {
-    const record = writeStoredRecord(token, expiresAt);
-    return {
-      action: "write",
-      command_id: commandId,
-      status: "written",
-      expires_at: record.expires_at || "",
-    };
+  function saveResult(commandId, value) {
+    writeJson(RESULT_PREFIX + commandId, { saved_at:Date.now(), value });
   }
 
-  function handleDelete(commandId) {
-    window.localStorage.removeItem(STORAGE_KEY);
-    deleteLegacyCookie();
-    const deleted = window.localStorage.getItem(STORAGE_KEY) === null;
-    return {
-      action: "delete",
-      command_id: commandId,
-      status: deleted ? "deleted" : "error",
-      error: deleted ? "" : "delete_verification_failed",
-    };
-  }
-
-  function onRender(event) {
-    if (!event || event.source !== window.parent) {
-      return;
-    }
-    const message = event.data || {};
-    if (message.type !== "streamlit:render") {
-      return;
-    }
+  async function handleEdgeAction(action, commandId, args) {
+    const path = ACTION_PATHS[action];
+    if (!path) return { ok:false, error:"unsupported_edge_action", action, command_id:commandId };
+    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 cached = cachedResult(commandId);
+    if (cached) return cached;
+    const result = Object.assign(await postJson(apiOrigin, path, payload), { action, command_id:commandId });
+    // Cache only confirmed successes. Transient network and HTTP failures must
+    // be retryable on the next Streamlit rerun with the same idempotency key.
+    if (result.ok) saveResult(commandId, result);
+
+    return result;
+  }
 
+  async function onRender(event) {
+    if (!event || event.source !== window.parent) return;
+    const message = event.data || {};
+    if (message.type !== "streamlit:render") return;
     const args = message.args || {};
     const action = typeof args.action === "string" ? args.action : "read";
-    const commandId =
-      typeof args.command_id === "string" ? args.command_id : "";
-
+    const commandId = typeof args.command_id === "string" ? args.command_id : "";
     let response;
     try {
-      if (action === "write") {
-        response = handleWrite(
-          commandId,
-          typeof args.token === "string" ? args.token : "",
-          typeof args.expires_at === "string" ? args.expires_at : ""
-        );
-      } else if (action === "delete") {
-        response = handleDelete(commandId);
-      } else {
-        response = handleRead(commandId);
-      }
+      response = ["read","write","delete"].includes(action)
+        ? handleRestore(action, commandId, args)
+        : await handleEdgeAction(action, commandId, args);
     } catch (_) {
-      response = {
-        action: action,
-        command_id: commandId,
-        status: "error",
-        error: "storage_unavailable",
-      };
+      response = { ok:false, action, command_id:commandId, status:"error", error:"browser_bridge_failure" };
     }
-
-    setFrameHeight(0);
+    setFrameHeight();
     sendValue(response);
   }
 
   window.addEventListener("message", onRender);
-  sendMessage("streamlit:componentReady", { apiVersion: 1 });
-  setFrameHeight(0);
+  sendMessage("streamlit:componentReady", { apiVersion:1 });
+  setFrameHeight();
 })();
 </script>
 </body>

task_executor.py

@@ -0,0 +1,447 @@
+# -*- coding: utf-8 -*-
+"""SmartPaper R3.1 independent task executor.
+
+Bootstrap is intentionally lightweight: the child process validates the local
+request, claims the D1 task, writes a local claim handshake, and only then
+imports the heavy paper-generation modules.  A Streamlit rerun/browser event
+therefore cannot be mistaken for successful executor startup, and import or
+pipeline failures after claim can be finalized through the authoritative
+Worker so quota is never left reserved by an unowned task.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import os
+import re
+import sys
+import threading
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from urllib.parse import urlsplit
+
+import requests
+
+from config import APP_CONFIG
+
+_log_handlers = [logging.StreamHandler(sys.stdout)]
+_executor_log_path = str(os.environ.get("SMARTPAPER_EXECUTOR_LOG_PATH") or "").strip()
+if _executor_log_path:
+    try:
+        Path(_executor_log_path).parent.mkdir(parents=True, exist_ok=True)
+        _log_handlers.append(logging.FileHandler(_executor_log_path, encoding="utf-8"))
+    except Exception:
+        pass
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    handlers=_log_handlers,
+)
+
+
+CLAIM_HANDSHAKE_FILENAME = "executor_claim.json"
+
+
+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
+
+
+def _atomic_json(path: Path, value: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
+    os.replace(temporary, path)
+
+
+def _claim_marker_path(task_dir: Path) -> Path:
+    return task_dir / CLAIM_HANDSHAKE_FILENAME
+
+
+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:
+    """Classify only post-claim execution failures.
+
+    This deliberately avoids importing the generation pipeline at module import
+    time.  DeepSeek exception class names are stable public names inside this
+    project and are inspected only after claim/bootstrap has succeeded.
+    """
+    if isinstance(exc, ExecutorApiError):
+        return exc.status_code == 429 or exc.status_code >= 500
+    class_name = exc.__class__.__name__
+    if class_name in {"DeepSeekAuthenticationError", "DeepSeekBalanceError", "DeepSeekRequestError"}:
+        return False
+    if isinstance(exc, (requests.exceptions.Timeout, requests.exceptions.ConnectionError)):
+        return True
+    if class_name == "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.1-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))
+        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:
+                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 _bootstrap_claim(*, task_dir: Path, task_id: str, user_id: int, request_data: dict, api: ExecutorApi) -> tuple[dict, int, str]:
+    """Claim first and publish a PID-bound local handshake before heavy imports."""
+    request_hash = str(request_data.get("request_hash") or "").strip()
+    if not request_hash:
+        raise RuntimeError("request_hash_missing")
+    claim = api.claim(task_id=task_id, user_id=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")
+    try:
+        _atomic_json(
+            _claim_marker_path(task_dir),
+            {
+                "task_id": task_id,
+                "user_id": int(user_id),
+                "request_hash": request_hash,
+                "pid": os.getpid(),
+                "execution_epoch": epoch,
+                "executor_instance_id": instance_id,
+                "claimed_at": datetime.now(timezone.utc).isoformat(),
+            },
+        )
+    except Exception as exc:
+        try:
+            api.finish(
+                {
+                    "task_id": task_id,
+                    "user_id": int(user_id),
+                    "execution_epoch": epoch,
+                    "executor_instance_id": instance_id,
+                    "status": "failed",
+                    "safe_error_code": "E_EXECUTOR_HANDSHAKE_FAILED",
+                    "safe_error_detail": _safe(exc, 800),
+                    "current_node": "executor_handshake_failed",
+                }
+            )
+        finally:
+            raise
+    return task, epoch, instance_id
+
+
+def _finish_bootstrap_failure(api: ExecutorApi, *, task_id: str, user_id: int, epoch: int, instance_id: str, exc: Exception) -> None:
+    try:
+        api.finish(
+            {
+                "task_id": task_id,
+                "user_id": user_id,
+                "execution_epoch": epoch,
+                "executor_instance_id": instance_id,
+                "status": "failed",
+                "safe_error_code": "E_EXECUTOR_BOOTSTRAP_FAILED",
+                "safe_error_detail": _safe(exc, 800),
+                "current_node": "executor_bootstrap_failed",
+            }
+        )
+    except Exception as finish_exc:
+        logging.warning("[SP_EXECUTOR] bootstrap_finalize_rejected task_id=%s error=%s", task_id, _safe(finish_exc, 220))
+
+
+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")
+
+    api = ExecutorApi()
+    task, epoch, instance_id = _bootstrap_claim(
+        task_dir=task_dir,
+        task_id=args.task_id,
+        user_id=args.user_id,
+        request_data=request_data,
+        api=api,
+    )
+    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])
+
+    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:
+        # Heavy paper modules are deliberately imported only after D1 claim and
+        # the local claim handshake. Any import/init failure is now observable
+        # and can atomically release the reserved quota.
+        from core.generation_runner import GenerationRunner
+
+        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")
+        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 isinstance(exc, (ImportError, ModuleNotFoundError)):
+            logging.exception("[SP_EXECUTOR] bootstrap_import_failed task_id=%s epoch=%s error=%s", args.task_id, epoch, _safe(exc, 240))
+            _finish_bootstrap_failure(api, task_id=args.task_id, user_id=args.user_id, epoch=epoch, instance_id=instance_id, exc=exc)
+            return 4
+        if _is_recoverable_generation_failure(exc):
+            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:
+            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/desktop.ini

@@ -0,0 +1,2 @@
+[.ShellClassInfo]
+IconResource=C:\Program Files\Google\Drive File Stream\129.0.1.0\GoogleDriveFS.exe,27

utils/docx_builder.py

@@ -3,9 +3,14 @@ import io
 import mistune
 from bs4 import BeautifulSoup
 from docx import Document
-from docx.shared import Inches, Pt
+from docx.shared import Pt
 from config import APP_CONFIG
 
+
+class DocxBuildError(RuntimeError):
+    """Raised when markdown cannot be rendered into a valid DOCX document."""
+
+
 class DocxRenderer(mistune.HTMLRenderer):
     def __init__(self):
         super().__init__()
@@ -130,17 +135,18 @@ class DocxRenderer(mistune.HTMLRenderer):
         return ''
 
 def text_to_docx(markdown_content):
+    """Render markdown strictly; never disguise renderer failures as success."""
     try:
         renderer = DocxRenderer()
-        mistune.create_markdown(renderer=renderer, plugins=['table'])(markdown_content)
+        mistune.create_markdown(renderer=renderer, plugins=['table'])(str(markdown_content or ''))
         doc = renderer.doc
-    except Exception:
-        doc = Document()
-        for line in markdown_content.split('\n'):
-            if line.strip(): 
-                doc.add_paragraph(line)
-                
-    stream = io.BytesIO()
-    doc.save(stream)
-    stream.seek(0)
-    return stream
+        if not doc.paragraphs and not doc.tables:
+            raise DocxBuildError("DOCX renderer produced no document content.")
+        stream = io.BytesIO()
+        doc.save(stream)
+        stream.seek(0)
+        return stream
+    except DocxBuildError:
+        raise
+    except Exception as exc:
+        raise DocxBuildError(f"DOCX rendering failed: {exc}") from exc

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