Deploy SmartPaper v1.1 R2.1Fix (#4)

- Deploy SmartPaper v1.1 R2.1Fix (6069e4f7e41958457729b3868c051c3c61dc8469)

app.py

@@ -20,6 +20,8 @@ warnings.filterwarnings("ignore", message="coroutine 'expire_cache' was never aw
 import os
 import streamlit.components.v1 as components
 import time
+import logging
+import threading
 import re
 import math
 import json
@@ -43,6 +45,11 @@ from i18n import LANGUAGE_OPTIONS, get_amj_csv_template, get_asq_txt_template, g
 os.environ["STREAMLIT_SERVER_ENABLE_CORS"] = "false"
 os.environ["STREAMLIT_SERVER_ENABLE_XSRF_PROTECTION"] = "false"
 
+# Streamlit executes the app on its ScriptRunner thread rather than Python's
+# process main thread. Capture that thread so background worker threads cannot
+# emit browser/D1 heartbeats on behalf of work they do not own.
+STREAMLIT_SCRIPT_THREAD_ID = threading.get_ident()
+
 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")
@@ -52,7 +59,9 @@ META_SECTIONS_BLACKLIST = set(APP_CONFIG["blacklists"]["meta_sections"])
 
 
 def update_heartbeat():
-    """Refresh the authoritative Cloudflare/D1 task lease only."""
+    """Refresh D1 only while the Python generation process is actually active."""
+    if threading.get_ident() != STREAMLIT_SCRIPT_THREAD_ID:
+        return
     now = time.time()
     task = st.session_state.get("sp_generation_task") or {}
     task_id = str(task.get("task_id") or "").strip()
@@ -66,10 +75,12 @@ def update_heartbeat():
             current_node=str(st.session_state.get("sp_current_node") or "")[:240],
         )
         st.session_state["sp_last_task_heartbeat"] = now
-    except Exception:
-        # The hidden browser bridge continues heartbeats independently. A single
-        # transient UI heartbeat must not interrupt model generation.
-        pass
+    except Exception as exc:
+        logging.warning(
+            "[SP_TASK] heartbeat_enqueue_failed task_id=%s error=%s",
+            task_id,
+            type(exc).__name__,
+        )
 
 
 def is_system_busy():
@@ -407,6 +418,12 @@ def _begin_managed_generation_task(*, is_resume: bool, loaded_state: dict | None
     st.session_state["sp_generation_idempotency_key"] = idempotency_key
     st.session_state["sp_generation_task"] = task
     st.session_state["sp_last_task_heartbeat"] = time.time()
+    logging.info(
+        "[SP_TASK] task_started task_id=%s mode=%s language=%s",
+        str(task.get("task_id") or ""),
+        mode,
+        language,
+    )
     return task, idempotency_key
 
 
@@ -530,6 +547,12 @@ def _process_pending_task_finalization():
         if not package_bytes:
             raise RuntimeError("validated_package_file_empty")
 
+    logging.info(
+        "[SP_TASK] task_finalization_request task_id=%s status=%s node=%s",
+        task_id,
+        status,
+        str(pending.get("current_node") or "")[:120],
+    )
     response = finish_generation_task(
         task_id,
         status=status,
@@ -538,6 +561,13 @@ def _process_pending_task_finalization():
         current_node=str(pending.get("current_node") or "")[:240],
     )
 
+    logging.info(
+        "[SP_TASK] task_finalization_confirmed task_id=%s status=%s reservation=%s",
+        task_id,
+        status,
+        str((response.get("reservation") or {}).get("status") or ""),
+    )
+
     if status == "completed":
         filename = str(package.get("filename") or "SmartPaper.zip")
         pending_path = str(package.get("pending_path") or "")
@@ -863,7 +893,12 @@ progress_bar = st.empty()
 monitor_ui = st.empty()
 
 def render_monitor(agent, phase, current_sec="", details=""):
-    st.session_state["sp_current_node"] = str(current_sec or phase or "")[:240]
+    node = str(current_sec or phase or "")[:240]
+    st.session_state["sp_current_node"] = node
+    if node and node != st.session_state.get("sp_last_logged_node"):
+        task_id = str((st.session_state.get("sp_generation_task") or {}).get("task_id") or "")
+        logging.info("[SP_TASK] node_enter task_id=%s node=%s", task_id, node)
+        st.session_state["sp_last_logged_node"] = node
     monitor_ui.markdown(f"""
     <div class='log-box'>
         <div><b>{tr(lang_code, 'monitor_pipeline')}:</b> {phase}</div><div><b>{tr(lang_code, 'monitor_node')}:</b> {current_sec}</div><div><b>{tr(lang_code, 'monitor_log')}:</b> {details}</div><hr>
@@ -922,6 +957,7 @@ if st.session_state.start_pipeline or st.session_state.resume_pipeline:
 
     st.session_state.is_running = True
     agent = SSciPipelineAgent(global_ds_key, global_s2_key)
+    agent.set_execution_progress_hook(update_heartbeat)
     SAFE_CHUNK_WORDS = agent.chunk_words_default
     qual_engine = get_qualitative_engine(agent)
     quant_engine = get_quantitative_engine()
@@ -1355,6 +1391,13 @@ Output only final academic prose. Do not report the word count."""
             render_monitor(agent, tr(lang_code, "complete"), "✅", tr(lang_code, "document_generated"))
             
     except Exception as e:
+        task_id = str((st.session_state.get("sp_generation_task") or {}).get("task_id") or "")
+        logging.exception(
+            "[SP_TASK] generation_exception task_id=%s node=%s error_type=%s",
+            task_id,
+            str(st.session_state.get("sp_current_node") or "generation")[:120],
+            type(e).__name__,
+        )
         if not completion_finalization_attempted:
             try:
                 _finish_managed_generation_task(
@@ -1368,6 +1411,12 @@ Output only final academic prose. Do not report the word count."""
                 pass
         st.error(tr(lang_code, "startup_error", error=e))
     finally:
+        task_id = str((st.session_state.get("sp_generation_task") or {}).get("task_id") or "")
+        logging.info(
+            "[SP_TASK] pipeline_exit task_id=%s node=%s",
+            task_id,
+            str(st.session_state.get("sp_current_node") or "")[:120],
+        )
         clear_heartbeat()
         st.session_state.is_running = False
         st.session_state.start_pipeline = False

config.py

@@ -21,11 +21,18 @@ def load_config():
         "rag_chunk_size": int(os.environ.get("RAG_CHUNK_SIZE", "3000")),
         "rag_top_k": int(os.environ.get("RAG_TOP_K", "3")),
         "max_extract_chars": int(os.environ.get("MAX_EXTRACT_CHARS", "150000")),
+        # A DeepSeek request must fail before the 90-second Worker generation
+        # lease can expire without real execution progress. Retries renew the
+        # lease through the execution heartbeat hook.
         "api_timeouts": {
-            "standard": (15, 180),
-            "stream": (15, 300),
+            "standard": (15, 70),
+            "stream": (15, 70),
             "search": 15
         },
+        "deepseek_stream_idle_seconds": max(
+            20,
+            min(65, int(os.environ.get("DEEPSEEK_STREAM_IDLE_SECONDS", "60"))),
+        ),
         
         # --- API 节点配置 ---
         "deepseek_api_url": os.environ.get(

core/pipeline_agent.py

@@ -96,6 +96,7 @@ class SSciPipelineAgent:
             for source in ("s2", "oa")
         }
         self._cancel_checker = None
+        self._execution_progress_hook = None
         self.s2_fused = False
         self.oa_fused = False
         self.citation_map = {}
@@ -147,6 +148,10 @@ class SSciPipelineAgent:
         self.openalex_api_url = APP_CONFIG["openalex_api_url"]
         self.s2_api_url = APP_CONFIG["s2_api_url"]
         self.timeouts = APP_CONFIG["api_timeouts"]
+        self.deepseek_stream_idle_seconds = max(
+            20,
+            int(APP_CONFIG.get("deepseek_stream_idle_seconds", 60)),
+        )
         
         self.global_memory = {
             "hypotheses": "", "variables": "", "model_summary": "",
@@ -164,6 +169,23 @@ class SSciPipelineAgent:
     def set_cancel_checker(self, checker):
         self._cancel_checker = checker
 
+    def set_execution_progress_hook(self, hook):
+        """Register a best-effort callback tied to real backend execution."""
+        self._execution_progress_hook = hook
+
+    def _touch_execution_progress(self, stage=""):
+        hook = self._execution_progress_hook
+        if hook is None:
+            return
+        try:
+            hook()
+        except Exception as exc:
+            logging.warning(
+                "Execution heartbeat hook failed | stage=%s | error=%s",
+                str(stage or "unknown")[:80],
+                type(exc).__name__,
+            )
+
     def _check_cancelled(self):
         if self._cancel_checker is not None:
             result = self._cancel_checker()
@@ -708,6 +730,13 @@ class SSciPipelineAgent:
 
             try:
                 self._check_cancelled()
+                self._touch_execution_progress("deepseek_static_request")
+                logging.info(
+                    "DeepSeek static attempt start | attempt=%s/%s | model=%s",
+                    attempt + 1,
+                    self.deepseek_max_retries,
+                    current_model,
+                )
                 response = requests.post(
                     self.deepseek_api_url,
                     headers=self._deepseek_headers(self.ds_key),
@@ -716,6 +745,7 @@ class SSciPipelineAgent:
                 )
 
                 self._check_cancelled()
+                self._touch_execution_progress("deepseek_static_response")
                 if response.status_code != 200:
                     self._raise_for_deepseek_status(response)
 
@@ -769,6 +799,7 @@ class SSciPipelineAgent:
                     self.deepseek_max_retries,
                     exc,
                 )
+                self._touch_execution_progress("deepseek_static_retry")
                 if is_last:
                     break
                 time.sleep(self._retry_delay(attempt))
@@ -844,10 +875,22 @@ class SSciPipelineAgent:
             emitted_content = False
             reasoning_open = False
             reasoning_chunks = 0
+            content_chunks = 0
+            content_chars = 0
+            first_content_at = None
             start_time = time.time()
+            last_model_activity = start_time
 
             try:
                 self._check_cancelled()
+                self._touch_execution_progress("deepseek_stream_request")
+                logging.info(
+                    "DeepSeek stream attempt start | attempt=%s/%s | model=%s | idle_limit=%ss",
+                    attempt + 1,
+                    self.deepseek_max_retries,
+                    current_model,
+                    self.deepseek_stream_idle_seconds,
+                )
                 with requests.post(
                     self.deepseek_api_url,
                     headers=self._deepseek_headers(self.ds_key),
@@ -863,9 +906,22 @@ class SSciPipelineAgent:
                     ):
                         self._check_cancelled()
                         event = self._parse_sse_line(raw_line)
+                        now = time.time()
+                        if now - last_model_activity > self.deepseek_stream_idle_seconds:
+                            raise DeepSeekAPIError(
+                                "DeepSeek stream produced no model progress for "
+                                f"{self.deepseek_stream_idle_seconds} seconds."
+                            )
                         if not event:
                             continue
                         if event.get("done"):
+                            logging.info(
+                                "DeepSeek stream done marker | seconds=%.1f | reasoning_chunks=%s | content_chunks=%s | content_chars=%s",
+                                now - start_time,
+                                reasoning_chunks,
+                                content_chunks,
+                                content_chars,
+                            )
                             break
 
                         choices = event.get("choices") or []
@@ -879,7 +935,9 @@ class SSciPipelineAgent:
                         content = delta.get("content") or ""
 
                         if reasoning_content:
+                            last_model_activity = time.time()
                             reasoning_chunks += 1
+                            self._touch_execution_progress("deepseek_reasoning")
                             if not reasoning_open:
                                 reasoning_open = True
                                 logging.info(
@@ -902,6 +960,16 @@ class SSciPipelineAgent:
                                 )
 
                         if content:
+                            last_model_activity = time.time()
+                            self._touch_execution_progress("deepseek_content")
+                            if first_content_at is None:
+                                first_content_at = last_model_activity
+                                logging.info(
+                                    "DeepSeek first content | seconds=%.1f",
+                                    first_content_at - start_time,
+                                )
+                            content_chunks += 1
+                            content_chars += len(str(content))
                             if (
                                 reasoning_open
                                 and self.deepseek_show_reasoning
@@ -919,9 +987,13 @@ class SSciPipelineAgent:
                         "DeepSeek stream ended without final content."
                     )
 
+                self._touch_execution_progress("deepseek_stream_complete")
                 logging.info(
-                    "DeepSeek stream complete | seconds=%.1f",
+                    "DeepSeek stream complete | seconds=%.1f | reasoning_chunks=%s | content_chunks=%s | content_chars=%s",
                     time.time() - start_time,
+                    reasoning_chunks,
+                    content_chunks,
+                    content_chars,
                 )
                 return
 
@@ -953,6 +1025,7 @@ class SSciPipelineAgent:
                         "content had already been emitted."
                     ) from exc
 
+                self._touch_execution_progress("deepseek_stream_retry")
                 is_last = attempt == self.deepseek_max_retries - 1
                 if is_last:
                     break

managed_auth.py

@@ -29,6 +29,31 @@ from smartpaper_restore_bridge import edge_bridge
 EDGE_ASSERTION_COOKIE_NAME = "smartpaper_edge_assertion"
 EDGE_ASSERTION_AUDIENCE = "smartpaper-streamlit"
 ASSERTION_CLOCK_SKEW_SECONDS = 30
+EXECUTION_HEARTBEAT_PROOF_VERSION = "sp-execution-heartbeat-v1"
+
+
+def _execution_heartbeat_proof(task_id: str, current_node: str) -> tuple[int, str]:
+    """Create a short-lived HMAC proof that only the HF backend can mint."""
+    user = st.session_state.get("sp_user") or {}
+    user_id = str(user.get("id") or "").strip()
+    normalized_task = str(task_id or "").strip()
+    normalized_node = str(current_node or "")[:240]
+    if not user_id or not normalized_task:
+        raise RuntimeError("execution_heartbeat_identity_required")
+    issued_at = int(time.time())
+    task_hash = hashlib.sha256(normalized_task.encode("utf-8")).hexdigest()
+    user_hash = hashlib.sha256(user_id.encode("utf-8")).hexdigest()
+    node_hash = hashlib.sha256(normalized_node.encode("utf-8")).hexdigest()
+    canonical = (
+        f"{EXECUTION_HEARTBEAT_PROOF_VERSION}:"
+        f"{task_hash}:{user_hash}:{node_hash}:{issued_at}"
+    )
+    signature = hmac.new(
+        _service_key().encode("utf-8"),
+        canonical.encode("ascii"),
+        hashlib.sha256,
+    ).hexdigest()
+    return issued_at, signature
 
 
 def _runtime_value(name: str) -> str:
@@ -282,11 +307,19 @@ def heartbeat_generation_task(task_id: str, *, current_node: str = "") -> dict[s
     normalized = str(task_id or "").strip()
     if not normalized:
         raise RuntimeError("task_id_required")
+    normalized_node = str(current_node or "")[:240]
+    issued_at, proof = _execution_heartbeat_proof(normalized, normalized_node)
     bucket = int(time.time() // 20)
     _edge_call(
         "task_heartbeat",
         f"heartbeat:{normalized}:{bucket}",
-        {"task_id": normalized, "current_node": str(current_node or "")[:240]},
+        {
+            "task_id": normalized,
+            "current_node": normalized_node,
+            "heartbeat_source": "execution",
+            "heartbeat_issued_at": issued_at,
+            "heartbeat_proof": proof,
+        },
         fire_and_forget=True,
         key_prefix="sp_task_heartbeat",
     )

smartpaper_restore_bridge/frontend/index.html

@@ -199,25 +199,14 @@
     }
   }
 
-  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);
-    }
-  }
-
   function startBackgroundWork() {
+    // R2: the browser may retry commands that Python already emitted, but it
+    // must never manufacture task liveness merely because the tab remains open.
+    // Only real backend execution progress is allowed to enqueue task_heartbeat.
     if (heartbeatTimer) return;
     heartbeatTimer = window.setInterval(() => {
-      heartbeatActiveTask();
       drainPending();
     }, HEARTBEAT_MS);
-    heartbeatActiveTask();
     drainPending();
   }