Deploy SmartPaper production v1.1 R1Fix clean

HF_COMMIT_SHA.txt

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

app.py

@@ -3,6 +3,8 @@
 from core.meta_extractor import extract_golden_metadata
 from core.rag_engine import initialize_rag
 from core.qualitative_engine import get_qualitative_engine
+from core.quantitative_engine import get_quantitative_engine
+from core.production_guard import build_validated_package, replace_abstract
 import streamlit as st
 
 st.set_page_config(
@@ -19,21 +21,14 @@ 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 config import APP_CONFIG, AMJ_PROMPTS, ASQ_PROMPTS
+from utils.file_parser import extract_text_from_files
 from core.pipeline_agent import SSciPipelineAgent
 from managed_auth import (
     require_alipay_login,
@@ -41,8 +36,9 @@ from managed_auth import (
     start_generation_task,
     heartbeat_generation_task,
     finish_generation_task,
+    render_payment_purchase,
 )
-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"
@@ -51,32 +47,19 @@ 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."""
+    """Refresh the authoritative Cloudflare/D1 task lease only."""
     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()
+    interval = max(3, int(APP_CONFIG.get("task_heartbeat_interval_seconds", 5)))
     last_remote = float(st.session_state.get("sp_last_task_heartbeat") or 0.0)
-    if not task_id or now - last_remote < 25:
+    if not task_id or now - last_remote < interval:
         return
-
     try:
         heartbeat_generation_task(
             task_id,
@@ -84,30 +67,23 @@ def update_heartbeat():
         )
         st.session_state["sp_last_task_heartbeat"] = now
     except Exception:
-        # A transient heartbeat failure must not interrupt paper generation.
+        # The hidden browser bridge continues heartbeats independently. A single
+        # transient UI heartbeat must not interrupt model 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
+    """D1 is the sole concurrency authority; no duplicate Hugging Face file lock."""
     return False
 
+
 def clear_heartbeat():
-    try:
-        if os.path.exists(HEARTBEAT_FILE):
-            os.remove(HEARTBEAT_FILE)
-    except:
-        pass
+    """Backward-compatible no-op after removal of the duplicate local lock."""
+    return None
+
 
 def stream_with_heartbeat(generator):
-    """包裹器：确保大模型长时间流式输出时，能够持续发送心跳，防止锁被释放"""
+    """Refresh the D1 lease while streaming model output."""
     last_beat = time.time()
     for chunk in generator:
         now = time.time()
@@ -115,7 +91,7 @@ def stream_with_heartbeat(generator):
             update_heartbeat()
             last_beat = now
         yield chunk
-# =====================================================================
+
 
 def clean_trailing_references(text, keywords=("reference", "references", "reference list", "参考文献")):
     lines = text.split('\n')
@@ -144,117 +120,6 @@ global_s2_key = _read_runtime_secret("S2_API_KEY")
 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."""
@@ -471,6 +336,9 @@ if "sp_generation_task" not in st.session_state: st.session_state.sp_generation_
 if "sp_generation_idempotency_key" not in st.session_state: st.session_state.sp_generation_idempotency_key = ""
 if "sp_last_task_heartbeat" not in st.session_state: st.session_state.sp_last_task_heartbeat = 0.0
 if "sp_current_node" not in st.session_state: st.session_state.sp_current_node = ""
+if "sp_pending_task_finalization" not in st.session_state: st.session_state.sp_pending_task_finalization = None
+if "sp_pending_package" not in st.session_state: st.session_state.sp_pending_package = None
+if "sp_terminal_notice" not in st.session_state: st.session_state.sp_terminal_notice = None
 
 
 def _safe_user_workspace_key(value):
@@ -511,6 +379,9 @@ def _prepare_new_generation_request() -> None:
     st.session_state["sp_current_node"] = "task_start"
     st.session_state["zip_buffer"] = None
     st.session_state["zip_filename"] = ""
+    st.session_state["sp_pending_task_finalization"] = None
+    st.session_state["sp_pending_package"] = None
+    st.session_state["sp_terminal_notice"] = None
 
 
 def _begin_managed_generation_task(*, is_resume: bool, loaded_state: dict | None, mode: str, language: str):
@@ -539,55 +410,200 @@ def _begin_managed_generation_task(*, is_resume: bool, loaded_state: dict | None
     return task, idempotency_key
 
 
+def _task_output_dir(task_id: str) -> str:
+    root = globals().get("_user_workspace") or os.path.join(WORKSPACE_DIR, "users", "anonymous")
+    path = os.path.join(root, "tasks", _safe_user_workspace_key(task_id), "outputs")
+    os.makedirs(path, exist_ok=True)
+    return path
+
+
+def _checkpoint_state_for_update() -> dict:
+    if not os.path.isfile(CHECKPOINT_FILE):
+        return {}
+    try:
+        with open(CHECKPOINT_FILE, "r", encoding="utf-8") as handle:
+            value = json.load(handle)
+        return value if isinstance(value, dict) else {}
+    except Exception:
+        return {}
+
+
+def _write_checkpoint_state(value: dict) -> None:
+    if not isinstance(value, dict) or not value:
+        return
+    os.makedirs(os.path.dirname(CHECKPOINT_FILE), exist_ok=True)
+    temporary = CHECKPOINT_FILE + ".tmp"
+    with open(temporary, "w", encoding="utf-8") as handle:
+        json.dump(value, handle, ensure_ascii=False)
+    os.replace(temporary, CHECKPOINT_FILE)
+
+
+def _persist_pending_finalization() -> None:
+    pending = st.session_state.get("sp_pending_task_finalization")
+    if not isinstance(pending, dict):
+        return
+    state = _checkpoint_state_for_update()
+    if not state:
+        return
+    state["pending_task_finalization"] = pending
+    package = st.session_state.get("sp_pending_package")
+    if isinstance(package, dict):
+        state["pending_package"] = package
+    _write_checkpoint_state(state)
+
+
+def _restore_pending_finalization() -> None:
+    if isinstance(st.session_state.get("sp_pending_task_finalization"), dict):
+        return
+    state = _checkpoint_state_for_update()
+    pending = state.get("pending_task_finalization") if isinstance(state, dict) else None
+    if not isinstance(pending, dict):
+        return
+    st.session_state["sp_pending_task_finalization"] = pending
+    package = state.get("pending_package")
+    if isinstance(package, dict):
+        st.session_state["sp_pending_package"] = package
+    task_id = str(pending.get("task_id") or "").strip()
+    if task_id and not st.session_state.get("sp_generation_task"):
+        st.session_state["sp_generation_task"] = {"task_id": task_id}
+
+
+def _remove_pending_checkpoint_metadata(*, delete_checkpoint=False, detach_task=False) -> None:
+    if delete_checkpoint:
+        try:
+            os.remove(CHECKPOINT_FILE)
+        except OSError:
+            pass
+        return
+    state = _checkpoint_state_for_update()
+    if not state:
+        return
+    state.pop("pending_task_finalization", None)
+    state.pop("pending_package", None)
+    if detach_task:
+        state["managed_task_idempotency_key"] = ""
+        state["managed_task_id"] = ""
+    _write_checkpoint_state(state)
+
+
+def _prune_user_output_packages(max_files: int = 5) -> None:
+    root = globals().get("_user_workspace")
+    if not root or not os.path.isdir(root):
+        return
+    packages = []
+    for current_root, _, names in os.walk(os.path.join(root, "tasks")):
+        for name in names:
+            if name.endswith(".zip") and not name.startswith(".pending_"):
+                path = os.path.join(current_root, name)
+                try:
+                    packages.append((os.path.getmtime(path), path))
+                except OSError:
+                    pass
+    packages.sort()
+    while len(packages) > max_files:
+        _, old_path = packages.pop(0)
+        try:
+            os.remove(old_path)
+        except OSError:
+            pass
+
+
+def _process_pending_task_finalization():
+    pending = st.session_state.get("sp_pending_task_finalization")
+    if not isinstance(pending, dict):
+        return None
+    task_id = str(pending.get("task_id") or "").strip()
+    status = str(pending.get("status") or "").strip().lower()
+    if not task_id or status not in {"completed", "failed", "cancelled", "resumable"}:
+        raise RuntimeError("invalid_pending_task_finalization")
+
+    package_bytes = None
+    package = st.session_state.get("sp_pending_package")
+    if status == "completed":
+        if not isinstance(package, dict):
+            raise RuntimeError("validated_package_metadata_missing")
+        pending_path = str(package.get("pending_path") or "").strip()
+        if not pending_path or not os.path.isfile(pending_path):
+            raise RuntimeError("validated_package_file_missing")
+        with open(pending_path, "rb") as handle:
+            package_bytes = handle.read()
+        if not package_bytes:
+            raise RuntimeError("validated_package_file_empty")
+
+    response = finish_generation_task(
+        task_id,
+        status=status,
+        safe_error_code=str(pending.get("safe_error_code") or "")[:80],
+        safe_error_detail=str(pending.get("safe_error_detail") or "")[:800],
+        current_node=str(pending.get("current_node") or "")[:240],
+    )
+
+    if status == "completed":
+        filename = str(package.get("filename") or "SmartPaper.zip")
+        pending_path = str(package.get("pending_path") or "")
+        final_path = os.path.join(_task_output_dir(task_id), os.path.basename(filename))
+        try:
+            os.replace(pending_path, final_path)
+        except OSError:
+            # The bytes were read and validated before quota commit, so the user
+            # can still download even if the local archival rename fails.
+            final_path = pending_path
+        st.session_state["zip_buffer"] = package_bytes
+        st.session_state["zip_filename"] = os.path.basename(filename)
+        st.session_state["sp_terminal_notice"] = {"kind": "success"}
+        _prune_user_output_packages()
+    elif status in {"failed", "cancelled"}:
+        st.session_state["sp_terminal_notice"] = {
+            "kind": "error" if status == "failed" else "warning",
+            "message": str(pending.get("safe_error_detail") or "")[:800],
+        }
+
+    _remove_pending_checkpoint_metadata(
+        delete_checkpoint=bool(pending.get("delete_checkpoint")),
+        detach_task=bool(pending.get("detach_checkpoint")),
+    )
+    if status == "resumable":
+        st.session_state["sp_generation_task"] = dict(response.get("task") or {"task_id": task_id})
+    else:
+        st.session_state["sp_generation_task"] = None
+        st.session_state["sp_generation_idempotency_key"] = ""
+        st.session_state["sp_last_task_heartbeat"] = 0.0
+    st.session_state["sp_pending_task_finalization"] = None
+    st.session_state["sp_pending_package"] = None
+    return response
+
+
 def _finish_managed_generation_task(
     status: str,
     *,
     safe_error_code: str = "",
     safe_error_detail: str = "",
     current_node: str = "",
+    delete_checkpoint: bool = False,
+    detach_checkpoint: bool = False,
 ):
     task = st.session_state.get("sp_generation_task") or {}
     task_id = str(task.get("task_id") or "").strip()
     if not task_id:
         return None
-
-    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}")
+    pending = {
+        "task_id": task_id,
+        "status": str(status or "").strip().lower(),
+        "safe_error_code": str(safe_error_code or "")[:80],
+        "safe_error_detail": str(safe_error_detail or "")[:800],
+        "current_node": str(current_node or "")[:240],
+        "delete_checkpoint": bool(delete_checkpoint),
+        "detach_checkpoint": bool(detach_checkpoint),
+    }
+    st.session_state["sp_pending_task_finalization"] = pending
+    _persist_pending_finalization()
+    return _process_pending_task_finalization()
 
 
 def _quota_start_error_message(language: str, error: Exception) -> str:
     raw = str(error or "")
-    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"])
+    return tr(language, "quota_exhausted") if "payment_required" in raw else raw
+
 
 inject_wechat_browser_notice()
 
@@ -617,6 +633,8 @@ _user_workspace = os.path.join(
 )
 os.makedirs(_user_workspace, exist_ok=True)
 CHECKPOINT_FILE = os.path.join(_user_workspace, "snapshot.json")
+_restore_pending_finalization()
+_process_pending_task_finalization()
 
 
 @_fragment
@@ -694,14 +712,16 @@ def render_workspace_controls(user):
                             safe_error_code="E_USER_CLEARED_CHECKPOINT",
                             safe_error_detail="User cleared the recoverable checkpoint.",
                             current_node="checkpoint_clear",
+                            delete_checkpoint=True,
                         )
                     except Exception as exc:
                         st.error(str(exc))
                         return
-                try:
-                    os.remove(CHECKPOINT_FILE)
-                except OSError:
-                    pass
+                else:
+                    try:
+                        os.remove(CHECKPOINT_FILE)
+                    except OSError:
+                        pass
                 clear_heartbeat()
                 st.rerun()
         st.markdown("---")
@@ -830,6 +850,15 @@ uploaded_interviews = st.session_state.get("up_asq_data")
 if st.session_state.get("sp_ui_blocked"):
     st.stop()
 
+if st.session_state.get("sp_payment_required") and not (st.session_state.start_pipeline or st.session_state.resume_pipeline):
+    render_payment_purchase(lang_code)
+
+_terminal_notice = st.session_state.pop("sp_terminal_notice", None)
+if isinstance(_terminal_notice, dict) and _terminal_notice.get("kind") in {"error", "warning"}:
+    _message = str(_terminal_notice.get("message") or "").strip()
+    if _message:
+        (st.error if _terminal_notice.get("kind") == "error" else st.warning)(_message)
+
 progress_bar = st.empty()
 monitor_ui = st.empty()
 
@@ -886,12 +915,16 @@ if st.session_state.start_pipeline or st.session_state.resume_pipeline:
         st.session_state.start_pipeline = False
         st.session_state.resume_pipeline = False
         st.error(_quota_start_error_message(lang_code, exc))
+        if "payment_required" in str(exc):
+            st.session_state["sp_payment_required"] = True
+            render_payment_purchase(lang_code)
         st.stop()
 
     st.session_state.is_running = True
     agent = SSciPipelineAgent(global_ds_key, global_s2_key)
-    SAFE_CHUNK_WORDS = agent.chunk_words_cyrillic if target_lang in ["Russian (Русский)", "Kazakh (Қазақша)"] else agent.chunk_words_default
+    SAFE_CHUNK_WORDS = agent.chunk_words_default
     qual_engine = get_qualitative_engine(agent)
+    quant_engine = get_quantitative_engine()
     completion_finalization_attempted = False
 
     try:
@@ -907,6 +940,7 @@ if st.session_state.start_pipeline or st.session_state.resume_pipeline:
             target_lang = state["target_lang"]
             words = state["words"]
             user_data_md = state["user_data_md"]
+            data_metadata = state.get("data_metadata", {})
             user_lit_context = state["user_lit_context"]
             
             agent.load_state(state["agent_state"])
@@ -931,6 +965,7 @@ if st.session_state.start_pipeline or st.session_state.resume_pipeline:
             pb = progress_bar.progress(1, text=tr(lang_code, "pipeline_start"))
             user_lit_context = ""
             user_data_md = ""
+            data_metadata = {}
             
             if st.session_state.run_mode == "amj":
                 if uploaded_papers_amj:
@@ -953,7 +988,7 @@ if st.session_state.start_pipeline or st.session_state.resume_pipeline:
                 
                 with st.spinner(tr(lang_code, "extract_quant")):
                     update_heartbeat()
-                    _, user_data_md, msg = resolve_empirical_data(uploaded_data_amj, lang_code)
+                    _, user_data_md, msg, data_metadata = quant_engine.resolve_empirical_data(uploaded_data_amj, lang_code)
                     st.info(msg)
                     
             elif st.session_state.run_mode == "asq":
@@ -977,12 +1012,12 @@ if st.session_state.start_pipeline or st.session_state.resume_pipeline:
                 
                 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"))
+                user_data_md, msg, data_metadata = qual_engine.resolve_qualitative_data(topic, keywords, raw_interviews)
+                st.info(msg or tr(lang_code, "qualitative_ready"))
 
             state = {
                 "topic": topic, "keywords": keywords, "target_lang": target_lang, "words": words,
-                "user_data_md": user_data_md, "user_lit_context": user_lit_context,
+                "user_data_md": user_data_md, "data_metadata": data_metadata, "user_lit_context": user_lit_context,
                 "completed_sections": [], "full_draft_en": "", "full_draft_translated": "",
                 "context_tail": "", "run_mode": st.session_state.run_mode, "interface_language": lang_code,
                 "managed_task_idempotency_key": managed_task_key,
@@ -1042,7 +1077,7 @@ if st.session_state.start_pipeline or st.session_state.resume_pipeline:
             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)))
+                anchor_response = ph.write_stream(stream_with_heartbeat(agent.stream_generate_anchors(topic, keywords, synthesis_result, refs_text, data_evidence=user_data_md, mode=st.session_state.run_mode)))
                 status.update(label=tr(lang_code, "anchor_done"), state="complete", expanded=False)
             
             # ==========================================================
@@ -1054,16 +1089,18 @@ if st.session_state.start_pipeline or st.session_state.resume_pipeline:
             conclusion_match = re.search(r'-+\s*CONCLUSION\s*-+\s*(.*?)$', anchor_response_clean, re.DOTALL | re.IGNORECASE)
             abstract_anchor = abstract_match.group(1).strip() if abstract_match else anchor_response_clean.strip()
             conclusion_anchor = conclusion_match.group(1).strip() if conclusion_match else ""
+            agent.global_memory["abstract_anchor"] = abstract_anchor
+            agent.global_memory["conclusion_anchor"] = conclusion_anchor
             # ==========================================================
             
             pb.progress(15, text=tr(lang_code, "outline_progress"))
             update_heartbeat()
-            plan = agent.generate_outline(topic, keywords, mode=st.session_state.run_mode)
+            plan = agent.generate_outline(topic, keywords, mode=st.session_state.run_mode, data_metadata=data_metadata)
             en_topic = str(plan.get("en_topic", topic))
             structure = plan.get("structure", [])
 
             en_abstract_clean = clean_markdown_headings(re.sub(r'(?i)^abstract[:\s]*', '', str(abstract_anchor)).strip())
-            trans_abstract_clean = clean_markdown_headings(re.sub(r'(?i)^(摘要|abstract|аннотация|түйіндеме)[:\s]*', '', agent.call_deepseek(_translation_system_prompt(target_lang), abstract_anchor, max_tokens=3000)).strip())
+            trans_abstract_clean = ""
 
             loc_abs = {"Simplified Chinese (简体中文)": "摘要", "Russian (Русский)": "Аннотация", "Kazakh (Қазақша)": "Түйіндеме"}.get(target_lang, "Abstract")
             loc_kw = {"Simplified Chinese (简体中文)": "关键词", "Russian (Русский)": "Ключевые слова", "Kazakh (Қазақша)": "Кілт сөздер"}.get(target_lang, "Keywords")
@@ -1071,21 +1108,10 @@ if st.session_state.start_pipeline or st.session_state.resume_pipeline:
             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})
-            
+            adaptive_structure = agent.build_generation_tasks(structure, int(words), SAFE_CHUNK_WORDS)
+            for task in adaptive_structure:
+                task["is_first"] = bool(task.get("section_first"))
+
             with st.spinner(tr(lang_code, "jit_status")):
                 update_heartbeat()
                 jit_texts = agent.prefetch_jit_literature(topic, keywords, adaptive_structure)
@@ -1112,7 +1138,7 @@ if st.session_state.start_pipeline or st.session_state.resume_pipeline:
         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}")
@@ -1123,360 +1149,220 @@ if st.session_state.start_pipeline or st.session_state.resume_pipeline:
             with col_trans:
                 st.markdown(tr(lang_code, "target_draft", flag=target_flag))
                 trans_ph = st.empty()
-                
-            if display_name in completed_sections: 
+            if display_name in completed_sections:
                 en_ph.success(tr(lang_code, "restored"))
-                trans_ph.success(tr(lang_code, "restored"))
-            else: 
+                trans_ph.info(tr(lang_code, "translation_waiting"))
+            else:
                 en_ph.info(tr(lang_code, "waiting"))
                 trans_ph.info(tr(lang_code, "translation_waiting"))
-                
             ui_blocks.append({"sec": sec, "en_ph": en_ph, "trans_ph": trans_ph})
 
         generation_aborted = False
         generation_failure_code = ""
         generation_failure_detail = ""
         accumulated_deviation = 0
-        remaining_sections = len(adaptive_structure)
+        remaining_sections = sum(1 for item in adaptive_structure if item["display_name"] not in completed_sections)
         all_cite_keys = list(agent.citation_map.keys())
-        valid_chunks = [s for s in adaptive_structure if not any(kw in s["section"].lower() for kw in ["abstract", "keyword", "摘要", "关键字", "appendix", "附录"])]
+        valid_chunks = [item for item in adaptive_structure if not any(word in item["section"].lower() for word in ["abstract","keyword","appendix"])]
         allocation_map = {}
         current_cite_idx = 0
-        
-        for idx, 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
+        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
 
-        translation_executor = ThreadPoolExecutor(max_workers=3)
-        previous_task = None
-
-        for i, block in enumerate(ui_blocks):
-            if generation_aborted: 
+        for block in ui_blocks:
+            if generation_aborted:
                 break
-                
             sec = block["sec"]
             en_ph = block["en_ph"]
             trans_ph = block["trans_ph"]
             sec_title = sec["section"]
             display_name = sec["display_name"]
-            
-            if display_name in completed_sections: 
-                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."
+            if display_name in completed_sections:
+                continue
+            remaining_sections = max(0, remaining_sections - 1)
+            adjustment = -accumulated_deviation / remaining_sections if remaining_sections else 0
+            adjusted_target = min(int(SAFE_CHUNK_WORDS * 1.5), max(80, int(sec["target_words"] + adjustment)))
+            is_quarantined = any(word in sec_title.lower() for word in ["abstract","keyword","conclusion","references","appendix"])
+            assigned_tags = ", ".join(allocation_map.get(display_name, [])) or "the single most relevant source"
+            citation_rule = "NO IN-TEXT CITATIONS ALLOWED IN THIS SECTION." if is_quarantined else f"Use claim-specific citation. Each sentence may contain at most one source tag. Candidate tags: {assigned_tags}. <AVAILABLE_CITATIONS>{refs_text}</AVAILABLE_CITATIONS>"
+            memory_context = agent._build_memory_context(sec_title, 2500)
+            memory_block = f"\n<GLOBAL_MEMORY>\n{memory_context}\n</GLOBAL_MEMORY>\n" if memory_context else ""
+            if getattr(st.session_state, "rag_engine", None) is not None:
+                snippets = st.session_state.rag_engine.get_relevant_context(display_name)
+                if snippets:
+                    memory_block += f"\n<TARGETED_RAG>\n{snippets}\n</TARGETED_RAG>\n"
+            subsection = sec.get("subsection")
+            heading_directive = "Do not output the section heading; it will be inserted by the system."
+            if subsection:
+                heading_directive += f" Do not output the subsection heading '{subsection}'; it will also be inserted by the system. Use level-3 headings only when analytically necessary."
+            elif sec_title.lower() in {"introduction", "conclusion"}:
+                heading_directive += " Do not use additional headings."
             else:
-                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() 
-            
+                heading_directive += " Use level-2 or level-3 headings only when they clarify the argument; never use level-4 headings or numbered headings."
+            table_rule = "Present statistical evidence in standard Markdown tables with plain-text captions." if sec["is_method_or_result"] and st.session_state.run_mode == "amj" else ""
+            required_content = str(sec.get("required_content") or "")
+            concepts = ", ".join(sec.get("concepts") or [])
+            qualitative_rule = f"Assigned qualitative concepts: {concepts}. Use only their assigned representative quotations and do not repeat a quotation already used elsewhere." if concepts else ""
+            data_injection = f"\n<RESEARCH_EVIDENCE>\n{user_data_md}\n</RESEARCH_EVIDENCE>\n" if user_data_md and sec["is_method_or_result"] else ""
+            completed_label = ", ".join(completed_sections)
+            anti_repeat = f"Already completed: {completed_label}. Do not restate their background, gap, or contribution." if completed_label else ""
+            sys_prompt = APP_CONFIG.get("sys_academic_writer", "") + "\n" + APP_CONFIG.get("deep_refinement_directive", "")
+            user_prompt = f"""Write the assigned English manuscript unit.
+Section: {sec_title}
+Subsection: {subsection or 'None'}
+Target length: {int(adjusted_target*0.9)}-{int(adjusted_target*1.1)} words.
+{heading_directive}
+{citation_rule}
+{table_rule}
+{required_content}
+{qualitative_rule}
+{anti_repeat}
+{memory_block}
+Literature synthesis: {synthesis_result}
+{data_injection}
+Previous tail: {context_tail[-2500:]}
+Output only final academic prose. Do not report the word count."""
+            if agent._count_tokens(sys_prompt) + agent._count_tokens(user_prompt) > 60000:
+                user_prompt = agent._truncate_context(user_prompt, 60000 - agent._count_tokens(sys_prompt))
+            render_monitor(agent, tr(lang_code, "drafting_phase", current=current_step+1, total=total_steps), display_name, tr(lang_code, "target_words", words=adjusted_target, engine=tr(lang_code, "engine_amj") if st.session_state.run_mode=="amj" else tr(lang_code, "engine_asq")))
             try:
                 with st.status(tr(lang_code, "draft_status", name=display_name), expanded=True) as status:
                     update_heartbeat()
-                    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()
-                    
+                    en_raw = en_ph.write_stream(stream_with_heartbeat(agent.stream_deepseek(sys_prompt, user_prompt, max_tokens=7000, model_role="writing", thinking=False, allow_fallback=False)))
                     status.update(label=tr(lang_code, "draft_done", name=display_name), state="complete", expanded=False)
-
-                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}"
-                
+                en_raw = re.sub(r'<think>.*?</think>', '', str(en_raw), flags=re.S).strip()
+                en_raw = re.sub(rf'(?im)^#\s*{re.escape(sec_title)}\s*\n+', '', en_raw).strip()
+                if subsection:
+                    en_raw = re.sub(rf'(?im)^##\s*{re.escape(subsection)}\s*\n+', '', en_raw).strip()
+                en_raw = re.sub(r'(?m)^(#{2,3})\s+[\d\.\s一二三四五六七八九十、]+\s*(.*)', r'\1 \2', en_raw)
+                en_raw = re.sub(r'(?m)^#{4,}\s+(.*)', r'**\1**', en_raw)
+                en_raw = re.sub(r'(?m)^#\s+(.+)$', r'## \1', en_raw)
                 if st.session_state.run_mode == "asq" and user_data_md and "appendix" not in sec_title.lower():
                     quote_dict = {tag: quote for quote, tag in re.findall(r'"([^"]+)"\s*\[(QUOTE-\d+)\]', user_data_md)}
                     for tag, quote in quote_dict.items():
-                        en_draft_clean = en_draft_clean.replace(tag, f'"{quote}"')
-                        
-                en_resolved = agent.resolve_citations(en_draft_clean, quarantine=is_quarantined)
+                        en_raw = en_raw.replace(tag, f'"{quote}"')
+                en_raw = agent.polish_academic_section(en_raw, display_name, abstract=False)
+                prefix = ""
+                if sec.get("section_first"):
+                    prefix += f"# {sec_title}\n\n"
+                if sec.get("subsection_first") and subsection:
+                    prefix += f"## {subsection}\n\n"
+                en_with_headings = prefix + en_raw
+                en_resolved = agent.audit_and_resolve_citations(en_with_headings, section_name=display_name, quarantine=is_quarantined)
                 en_ph.markdown(en_resolved)
-                
-                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))
-                
+                trans_ph.info(tr(lang_code, "translation_waiting"))
+                full_draft_en += en_resolved + "\n\n"
+                agent._extract_memory_from_text(en_raw, sec_title)
+                words_tail = en_raw.split()
+                context_tail = " ".join(words_tail[-200:])
+                actual_words = len(en_raw.split())
+                accumulated_deviation += actual_words - adjusted_target
                 current_step += 1
-                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:
+                completed_sections.append(display_name)
+                state.update({
+                    "completed_sections": completed_sections,
+                    "full_draft_en": full_draft_en,
+                    "full_draft_translated": "",
+                    "context_tail": context_tail,
+                    "agent_state": agent.export_state(),
+                    "data_metadata": data_metadata,
+                })
+                with open(CHECKPOINT_FILE, "w", encoding="utf-8") as checkpoint:
+                    json.dump(state, checkpoint, ensure_ascii=False)
+                st.caption(tr(lang_code, "word_count", actual=actual_words, target=adjusted_target, deviation=accumulated_deviation))
+                pb.progress(min(15 + int((current_step/max(1,total_steps))*75), 90), text=tr(lang_code, "section_complete_progress", name=display_name))
+            except Exception as error:
                 en_ph.error(tr(lang_code, "node_error"))
-                st.error(tr(lang_code, "node_error_detail", error=e))
+                st.error(tr(lang_code, "node_error_detail", error=error))
                 generation_aborted = True
                 generation_failure_code = "E_GENERATION_NODE_FAILED"
-                generation_failure_detail = str(e)[:800]
-                break
+                generation_failure_detail = str(error)[:800]
 
         if generation_aborted:
             _finish_managed_generation_task(
-                "resumable",
+                "failed",
                 safe_error_code=generation_failure_code or "E_GENERATION_INTERRUPTED",
-                safe_error_detail=generation_failure_detail or "Generation stopped with a recoverable checkpoint.",
+                safe_error_detail=generation_failure_detail or "Generation stopped; quota was released and the checkpoint was retained for a fresh retry.",
                 current_node=str(st.session_state.get("sp_current_node") or "generation"),
+                detach_checkpoint=True,
             )
 
-        if not generation_aborted and 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"))
+            pb.progress(91, 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
+            full_draft_en = clean_trailing_references(full_draft_en, ["Reference","References","Reference List"])
+            # The final abstract is derived from the completed English manuscript and frozen evidence.
+            final_abstract = agent.generate_final_abstract(topic, keywords, full_draft_en, data_evidence=user_data_md, mode=st.session_state.run_mode)
+            full_draft_en = replace_abstract(full_draft_en, final_abstract)
+            agent.validate_manuscript_structure(full_draft_en, adaptive_structure, translated=False)
+            if st.session_state.run_mode == "asq":
+                repeated_quotes = agent.find_repeated_qualitative_quotes(full_draft_en, data_metadata)
+                if repeated_quotes:
+                    raise RuntimeError(f"Repeated qualitative quotations: {repeated_quotes[:5]}")
+            # Only sources actually cited in the manuscript enter the reference list.
+            references = []
+            seen_refs = set()
+            for tag in sorted(agent.used_citations, key=agent._citation_tag_number):
+                ref = str((agent.citation_map.get(tag) or {}).get("ref") or "").strip()
+                if ref and ref not in seen_refs:
+                    references.append(f"- {ref}")
+                    seen_refs.add(ref)
+            if not references:
+                raise RuntimeError("No cited references remained after citation validation.")
+            full_draft_en = full_draft_en.rstrip() + "\n\n# References\n" + "\n\n".join(references)
+
+            pb.progress(94, text=tr(lang_code, "translation_running"))
+            update_heartbeat()
+            full_draft_translated = agent.translate_locked_manuscript(full_draft_en, target_lang)
+            agent.validate_translation_alignment(full_draft_en, full_draft_translated)
 
             bj_time = datetime.now(timezone(timedelta(hours=8))).strftime("%Y%m%d_%H%M")
-            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.
+            package = build_validated_package(
+                english_markdown=full_draft_en,
+                translated_markdown=full_draft_translated,
+                english_filename=f"Research_EN_{bj_time}.docx",
+                translated_filename=f"Research_{output_suffix}_{bj_time}.docx",
+                data_metadata=data_metadata,
+                mode=st.session_state.run_mode,
+                citation_audit=agent.citation_audit_log,
+                literature_metrics=agent.get_source_metrics(),
+            )
+            generated_zip_bytes = package.zip_bytes
+            generated_zip_filename = f"SmartPaper_{bj_time}.zip"
+
+            # Persist the already validated package in the current user's task
+            # directory before asking Worker/D1 to commit quota. The download is
+            # exposed only after the browser receives completed+committed.
+            task_id = str((st.session_state.get("sp_generation_task") or {}).get("task_id") or "").strip()
+            if not task_id:
+                raise RuntimeError("package_task_id_missing")
+            pending_path = os.path.join(_task_output_dir(task_id), ".pending_" + generated_zip_filename)
+            with open(pending_path, "wb") as output_file:
+                output_file.write(generated_zip_bytes)
+            st.session_state["sp_pending_package"] = {
+                "pending_path": pending_path,
+                "filename": generated_zip_filename,
+            }
             completion_finalization_attempted = True
             _finish_managed_generation_task(
                 "completed",
-                current_node="package_complete",
+                current_node="package_validated",
+                delete_checkpoint=True,
             )
-
-            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",
+                    "failed",
+                    safe_error_code="E_GENERATION_FAILED",
                     safe_error_detail=str(e)[:800],
                     current_node=str(st.session_state.get("sp_current_node") or "generation"),
+                    detach_checkpoint=True,
                 )
             except Exception:
                 pass

config.py

@@ -4,12 +4,18 @@ 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")),
+        "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")),
@@ -82,8 +88,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 +106,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,
@@ -116,6 +123,18 @@ def load_config():
             3,
             int(os.environ.get("SMARTPAPER_TASK_HEARTBEAT_SECONDS", "5")),
         ),
+        "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"),
         
@@ -331,7 +350,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/pipeline_agent.py

@@ -13,7 +13,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)])
@@ -71,6 +72,7 @@ class SSciPipelineAgent:
             self.openalex_key = st.secrets.get("OA_API_KEY", "")
         except: 
             self.openalex_key = os.environ.get("OA_API_KEY", "")
+        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,11 +84,14 @@ 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")
         }
@@ -95,14 +100,13 @@ class SSciPipelineAgent:
         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 {}
@@ -228,6 +232,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 +249,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: 
@@ -1347,15 +1401,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 +1453,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 +1594,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 +1692,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
+
+        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 expansion_phase in range(3):
+        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 +1993,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 +2120,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 +2186,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 +2241,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 +2256,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 +2271,580 @@ 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}.'
+            )
+        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

@@ -222,6 +222,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",

i18n.py

@@ -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": "将文件拖放到此处",
@@ -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",
@@ -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": "Перетащите файлы сюда",
@@ -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,43 +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:
@@ -61,15 +42,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 +66,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 +82,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 +105,45 @@ 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_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 +164,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 +171,215 @@ 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,
+    *,
+    fire_and_forget: bool = False,
+    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 {},
+        fire_and_forget=fire_and_forget,
+        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 _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)
+    return result
 
 
 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] 记录。当前未创建论文任务，也未预留额度。"
+    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(),
+        },
+        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)
+    return dict(result)
 
 
 def recover_generation_task() -> dict[str, Any]:
-    return _authorized_post("/task/recover", {})
+    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")
+    return dict(result)
 
 
 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})
+    normalized = str(task_id or "").strip()
+    if not normalized:
+        raise RuntimeError("task_id_required")
+    bucket = int(time.time() // 20)
+    _edge_call(
+        "task_heartbeat",
+        f"heartbeat:{normalized}:{bucket}",
+        {"task_id": normalized, "current_node": str(current_node or "")[:240]},
+        fire_and_forget=True,
+        key_prefix="sp_task_heartbeat",
+    )
+    return {"ok": True, "queued": True, "cancel_requested": False}
 
 
 def generation_task_status(task_id: str) -> dict[str, Any]:
-    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")
+    return dict(result)
+
+
+def finish_generation_task(
+    task_id: str,
+    *,
+    status: str,
+    safe_error_code: str = "",
+    safe_error_detail: str = "",
+    current_node: str = "",
+) -> dict[str, Any]:
+    """Finalize a task only after the browser receives Worker/D1 confirmation."""
+    normalized_task = str(task_id or "").strip()
+    normalized_status = str(status or "").strip().lower()
+    if not normalized_task:
+        raise RuntimeError("task_id_required")
+    if normalized_status not in {"completed", "failed", "cancelled", "resumable"}:
+        raise RuntimeError("invalid_finish_status")
+    command_id = f"finish:{normalized_task}:{normalized_status}"
+    result = _edge_call(
+        "task_finish",
+        command_id,
+        {
+            "task_id": normalized_task,
+            "status": normalized_status,
+            "safe_error_code": str(safe_error_code or "")[:80],
+            "safe_error_detail": str(safe_error_detail or "")[:800],
+            "current_node": str(current_node or "")[:240],
+        },
+        fire_and_forget=False,
+        key_prefix="sp_task_finish",
+    )
+    if result is None:
+        st.stop()
+    if not result.get("ok"):
+        _raise_edge_error(result, "task_finish_failed")
+    task = dict(result.get("task") or {})
+    reservation = dict(result.get("reservation") or {})
+    if str(task.get("task_id") or "") != normalized_task:
+        raise RuntimeError("task_finish_confirmation_mismatch")
+    if str(task.get("status") or "").lower() != normalized_status:
+        raise RuntimeError("task_finish_status_not_confirmed")
+    expected_reservation = {
+        "completed": "committed",
+        "failed": "released",
+        "cancelled": "released",
+        "resumable": "reserved",
+    }[normalized_status]
+    if str(reservation.get("status") or "").lower() != expected_reservation:
+        raise RuntimeError("task_quota_finalization_not_confirmed")
+    _merge_quota_into_session(result)
+    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 {},
+        fire_and_forget=False,
+        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 +391,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 +621,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,47 @@ 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={},
+        fire_and_forget=False,
+        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,
+    fire_and_forget: bool = False,
+    key: str = "sp_edge_bridge",
+) -> dict[str, Any] | None:
+    """Execute one authenticated same-origin Worker command in the browser.
+
+    The first synchronous invocation usually returns ``None``. Streamlit reruns
+    after the browser completes the request and then returns the result. Only
+    non-terminal heartbeat calls may use ``fire_and_forget=True``. Task
+    completion, failure, cancellation, payment, and account mutations remain
+    synchronous so Python receives the Worker/D1 postcondition.
+    """
+    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 {}),
+        fire_and_forget=bool(fire_and_forget),
         height=0,
         default=None,
         key=key,

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,62 @@
 (() => {
   "use strict";
 
-  const STORAGE_KEY = "smartpaper.restore_token.v1";
+  const RESTORE_KEY = "smartpaper.restore_token.v1";
   const LEGACY_COOKIE_NAME = "__Host-smartpaper_restore";
+  const ACTIVE_TASK_KEY = "smartpaper.edge.active_task.v3";
+  const PENDING_OPS_KEY = "smartpaper.edge.pending_ops.v3";
+  const RESULT_PREFIX = "smartpaper.edge.result.v3:";
   const MAX_TOKEN_LENGTH = 4096;
+  const MAX_RESULT_AGE_MS = 10 * 60 * 1000;
+  const HEARTBEAT_MS = 25000;
+  const REQUEST_TIMEOUT_MS = 20000;
   let lastResponseSignature = "";
+  let heartbeatTimer = null;
+
+  const ACTION_PATHS = Object.freeze({
+    task_start: "/edge/task/start",
+    task_heartbeat: "/edge/task/heartbeat",
+    task_finish: "/edge/task/finish",
+    task_cancel: "/edge/task/cancel",
+    task_status: "/edge/task/status",
+    task_recover: "/edge/task/recover",
+    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 +76,214 @@
   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;
+    const record = readJson(RESTORE_KEY, null);
+    const token = normalizedToken(record && record.token);
+    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, expires_at:expiresAt };
+  }
 
-    let record;
-    try {
-      record = JSON.parse(raw);
-    } catch (_) {
-      window.localStorage.removeItem(STORAGE_KEY);
-      return null;
-    }
+  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() };
+    if (!writeJson(RESTORE_KEY, record)) throw new Error("storage_unavailable");
+    const verified = readStoredRecord();
+    if (!verified || verified.token !== normalized) throw new Error("write_verification_failed");
+    deleteLegacyCookie();
+    return verified;
+  }
 
-    const token = normalizedToken(record && record.token);
-    if (!token) {
-      window.localStorage.removeItem(STORAGE_KEY);
-      return null;
+  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 || "" };
     }
-
-    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 (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" };
     }
-    if (expiryTimestamp !== null && expiryTimestamp <= Date.now()) {
-      window.localStorage.removeItem(STORAGE_KEY);
-      return null;
+    let record = readStoredRecord(); let source = "local_storage";
+    if (!record) {
+      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" };
+  }
+
+  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 ""; }
+  }
 
-    return { token: token, expires_at: expiresAt };
+  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 {
+        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); }
   }
 
-  function writeStoredRecord(token, expiresAt) {
-    const normalized = normalizedToken(token);
-    if (!normalized) {
-      throw new Error("invalid_token");
-    }
+  function pendingOperations() {
+    const value = readJson(PENDING_OPS_KEY, []);
+    return Array.isArray(value) ? value : [];
+  }
 
-    const record = {
-      token: normalized,
-      expires_at: typeof expiresAt === "string" ? expiresAt : "",
-      stored_at: new Date().toISOString(),
-    };
-    window.localStorage.setItem(STORAGE_KEY, JSON.stringify(record));
+  function storePending(items) { writeJson(PENDING_OPS_KEY, items.slice(-30)); }
 
-    const verified = readStoredRecord();
-    if (!verified || verified.token !== normalized) {
-      throw new Error("write_verification_failed");
+  function queuePending(operation) {
+    const items = pendingOperations();
+    if (!items.some(item => item.command_id === operation.command_id)) {
+      items.push(operation); storePending(items);
     }
-    deleteLegacyCookie();
-    return verified;
   }
 
-  function handleRead(commandId) {
-    let record = readStoredRecord();
-    let source = "local_storage";
+  function removePending(commandId) {
+    storePending(pendingOperations().filter(item => item.command_id !== commandId));
+  }
 
-    if (!record) {
-      const legacyToken = readLegacyCookie();
-      if (legacyToken) {
-        record = writeStoredRecord(legacyToken, "");
-        source = "legacy_cookie";
+  async function drainPending() {
+    for (const operation of pendingOperations()) {
+      const result = await postJson(operation.api_origin, operation.path, operation.payload);
+      if (result.ok) {
+        removePending(operation.command_id);
+      } else if ([401,403].includes(Number(result.http_status || 0))) {
+        // Authentication may be restored by a new Alipay session. Keep the
+        // transient heartbeat operation out of the terminal task state.
+        removePending(operation.command_id);
       }
     }
+  }
 
-    if (!record) {
-      return {
-        action: "read",
-        command_id: commandId,
-        status: "missing",
-      };
+  async function heartbeatActiveTask() {
+    const active = readJson(ACTIVE_TASK_KEY, null);
+    if (!active || !active.api_origin || !active.task_id) return;
+    const result = await postJson(active.api_origin, ACTION_PATHS.task_heartbeat, {
+      task_id:active.task_id,
+      current_node:active.current_node || "generation"
+    });
+    if (result.cancel_requested || [401,403,404,409].includes(Number(result.http_status || 0))) {
+      window.localStorage.removeItem(ACTIVE_TASK_KEY);
     }
+  }
 
-    return {
-      action: "read",
-      command_id: commandId,
-      status: "found",
-      token: record.token,
-      expires_at: record.expires_at || "",
-      source: source,
-    };
+  function startBackgroundWork() {
+    if (heartbeatTimer) return;
+    heartbeatTimer = window.setInterval(() => {
+      heartbeatActiveTask();
+      drainPending();
+    }, HEARTBEAT_MS);
+    heartbeatActiveTask();
+    drainPending();
   }
 
-  function handleWrite(commandId, token, expiresAt) {
-    const record = writeStoredRecord(token, expiresAt);
-    return {
-      action: "write",
-      command_id: commandId,
-      status: "written",
-      expires_at: record.expires_at || "",
-    };
+  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 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;
+  function saveResult(commandId, value) {
+    writeJson(RESULT_PREFIX + commandId, { saved_at:Date.now(), value });
+  }
+
+  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 fireAndForget = Boolean(args.fire_and_forget);
+
+    if (fireAndForget) {
+      queuePending({ command_id:commandId, api_origin:apiOrigin, path, payload });
+      drainPending();
+      return { ok:true, queued:true, action, command_id:commandId };
     }
-    const message = event.data || {};
-    if (message.type !== "streamlit:render") {
-      return;
+
+    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);
+
+    if (action === "task_start" && result.ok && result.task && result.task.task_id) {
+      writeJson(ACTIVE_TASK_KEY, { api_origin:apiOrigin, task_id:result.task.task_id, current_node:"generation" });
+    }
+    if (["task_finish", "task_cancel"].includes(action) && result.ok) {
+      window.localStorage.removeItem(ACTIVE_TASK_KEY);
     }
+    return result;
+  }
 
+  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();
+  startBackgroundWork();
 })();
 </script>
 </body>

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