Step 4A-6-C-R3.1 Executor Bootstrap and UI State Ownership Correction

.streamlit/desktop.ini

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

app.py

@@ -21,6 +21,7 @@ from config import APP_CONFIG
 from managed_auth import (
     cancel_generation_task,
     generation_task_status,
+    recover_generation_task,
     require_alipay_login,
     render_logged_in_account,
     render_payment_purchase,
@@ -265,6 +266,8 @@ for key, default in {
     "sp_pending_cancel_task_id": "",
     "sp_payment_required": False,
     "sp_terminal_notice": None,
+    "sp_task_recovery_initialized": False,
+    "sp_last_ui_task_status": "",
 }.items():
     if key not in st.session_state:
         st.session_state[key] = default
@@ -393,31 +396,54 @@ def _pid_is_task_executor(pid: int, task_id: str, task_dir: Path) -> bool:
 
 
 def _launch_executor(task_id: str, *, force: bool = False) -> int:
+    """Launch only after a PID-bound claim handshake is observed.
+
+    ``subprocess.Popen`` proves only that the OS accepted a child process. R3.1
+    waits until that exact child has successfully claimed the D1 task and
+    written ``executor_claim.json``. A child that exits/import-fails before
+    claim is therefore never reported as a running executor.
+    """
     task_dir = _task_dir(task_id)
     request_path = task_dir / "request.json"
     if not request_path.is_file():
         raise RuntimeError("task_request_manifest_missing")
     request_data = json.loads(request_path.read_text(encoding="utf-8"))
     numeric_id = int(request_data.get("user_id") or 0)
-    marker_path = task_dir / "executor_launch.json"
-    if marker_path.is_file():
+    request_hash = str(request_data.get("request_hash") or "").strip()
+    if numeric_id <= 0 or not request_hash:
+        raise RuntimeError("task_request_identity_invalid")
+
+    launch_marker_path = task_dir / "executor_launch.json"
+    claim_marker_path = task_dir / "executor_claim.json"
+    if launch_marker_path.is_file():
         try:
-            marker = json.loads(marker_path.read_text(encoding="utf-8"))
+            marker = json.loads(launch_marker_path.read_text(encoding="utf-8"))
             pid = int(marker.get("pid") or 0)
             verified_executor = _pid_is_task_executor(pid, task_id, task_dir)
             if verified_executor and not force:
-                return pid
+                claim_marker = {}
+                if claim_marker_path.is_file():
+                    try:
+                        claim_marker = json.loads(claim_marker_path.read_text(encoding="utf-8"))
+                    except Exception:
+                        claim_marker = {}
+                if int(claim_marker.get("pid") or 0) == pid and int(claim_marker.get("execution_epoch") or 0) > 0:
+                    return pid
             if verified_executor and force:
-                # Resume means D1 has already made the previous execution
-                # reclaimable. Stop only a PID whose /proc command line proves
-                # it is this task's old executor; a new epoch will be the sole
-                # authority after claim.
                 try:
                     os.kill(pid, signal.SIGTERM)
                 except OSError:
                     pass
         except Exception:
             pass
+
+    # A stale claim file must never satisfy a new launch/resume handshake.
+    try:
+        claim_marker_path.unlink(missing_ok=True)
+    except TypeError:
+        if claim_marker_path.exists():
+            claim_marker_path.unlink()
+
     log_path = task_dir / "executor.log"
     task_dir.mkdir(parents=True, exist_ok=True)
     command = [
@@ -427,20 +453,54 @@ def _launch_executor(task_id: str, *, force: bool = False) -> int:
         "--task-id", str(task_id),
         "--user-id", str(numeric_id),
     ]
-    with log_path.open("ab") as log_handle:
-        process = subprocess.Popen(
-            command,
-            cwd=str(Path(__file__).resolve().parent),
-            stdin=subprocess.DEVNULL,
-            stdout=log_handle,
-            stderr=subprocess.STDOUT,
-            close_fds=True,
-            start_new_session=True,
-            env=os.environ.copy(),
-        )
-    _atomic_json(marker_path, {"pid": process.pid, "launched_at": datetime.now(timezone.utc).isoformat()})
-    logging.info("[SP_UI] executor_launched task_id=%s pid=%s", task_id, process.pid)
-    return int(process.pid)
+    executor_env = os.environ.copy()
+    executor_env["SMARTPAPER_EXECUTOR_LOG_PATH"] = str(log_path.resolve())
+    process = subprocess.Popen(
+        command,
+        cwd=str(Path(__file__).resolve().parent),
+        stdin=subprocess.DEVNULL,
+        # Inherit stdout/stderr so executor bootstrap/errors are visible in HF
+        # Container Logs; task_executor also writes the same log to executor.log.
+        close_fds=True,
+        start_new_session=True,
+        env=executor_env,
+    )
+    _atomic_json(launch_marker_path, {"pid": process.pid, "launched_at": datetime.now(timezone.utc).isoformat()})
+    logging.info("[SP_UI] executor_spawned task_id=%s pid=%s", task_id, process.pid)
+
+    timeout_seconds = float(APP_CONFIG.get("executor_bootstrap_timeout_seconds", 20))
+    poll_seconds = max(0.05, float(APP_CONFIG.get("executor_bootstrap_poll_ms", 100)) / 1000.0)
+    deadline = time.monotonic() + timeout_seconds
+    while time.monotonic() < deadline:
+        if claim_marker_path.is_file():
+            try:
+                claim_marker = json.loads(claim_marker_path.read_text(encoding="utf-8"))
+            except Exception:
+                claim_marker = {}
+            if (
+                int(claim_marker.get("pid") or 0) == int(process.pid)
+                and str(claim_marker.get("task_id") or "") == str(task_id)
+                and str(claim_marker.get("request_hash") or "") == request_hash
+                and int(claim_marker.get("execution_epoch") or 0) > 0
+                and str(claim_marker.get("executor_instance_id") or "").strip()
+            ):
+                logging.info(
+                    "[SP_UI] executor_claim_confirmed task_id=%s pid=%s epoch=%s",
+                    task_id, process.pid, claim_marker.get("execution_epoch"),
+                )
+                return int(process.pid)
+        return_code = process.poll()
+        if return_code is not None:
+            logging.error("[SP_UI] executor_bootstrap_exit task_id=%s pid=%s rc=%s", task_id, process.pid, return_code)
+            raise RuntimeError(f"executor_bootstrap_failed_exit_{return_code}")
+        time.sleep(poll_seconds)
+
+    try:
+        process.terminate()
+    except Exception:
+        pass
+    logging.error("[SP_UI] executor_claim_timeout task_id=%s pid=%s", task_id, process.pid)
+    raise RuntimeError("executor_claim_timeout")
 
 
 def _process_pending_cancel() -> None:
@@ -502,6 +562,7 @@ def _process_pending_launch(lang_code: str) -> None:
         st.error("task_start_missing_task_id")
         return
     st.session_state["sp_generation_task"] = task
+    st.session_state["sp_last_ui_task_status"] = str(task.get("status") or "")
     try:
         request_path = _task_dir(task_id) / "request.json"
         if not request_path.is_file():
@@ -522,25 +583,19 @@ inject_wechat_browser_notice()
 DEFAULT_TOPIC = "Generative Artificial Intelligence in Ubiquitous Learning: Evaluating a Chatbot-based Recommendation Engine for Personalized and Context-aware Education"
 DEFAULT_KEYWORDS = "Ubiquitous computing; Context awareness; Deep learning; Personalized educational; Educational technologies; HCI; Recommendation system; LLMs; Generative AI; ChatGPT; Prompting engineering; Few-shot prompting"
 
-_fragment = getattr(st, "fragment", None) or getattr(st, "experimental_fragment", None)
-if _fragment is None:
-    def _fragment(function=None, **_kwargs):
-        if function is not None:
-            return function
-        return lambda fn: fn
-
-
 def _current_language_code() -> str:
     choice = st.session_state.get("interface_language_choice")
     return LANGUAGE_OPTIONS.get(choice, "zh")
 
 
 authenticated_user = require_alipay_login(_current_language_code())
+if not bool(st.session_state.get("sp_task_recovery_initialized")):
+    recover_generation_task()
+    st.session_state["sp_task_recovery_initialized"] = True
 _user_workspace = os.path.join(WORKSPACE_DIR, "users", _safe_key(authenticated_user.get("id")))
 os.makedirs(_user_workspace, exist_ok=True)
 
 
-@_fragment
 def render_workspace_controls(user):
     top_toolbar = st.container()
     with top_toolbar:
@@ -661,6 +716,7 @@ def _render_task_status_once() -> None:
     task_id = str(task.get("task_id") or "").strip()
     if not task_id:
         return
+    previous_status = str(st.session_state.get("sp_last_ui_task_status") or str(task.get("status") or ""))
     status = str(task.get("status") or "")
     if status in {"created", "running", "resumable", "cancelling"}:
         try:
@@ -670,6 +726,7 @@ def _render_task_status_once() -> None:
             status = str(task.get("status") or "")
         except Exception as exc:
             logging.warning("[SP_UI] status_poll_error task_id=%s error=%s", task_id, type(exc).__name__)
+    st.session_state["sp_last_ui_task_status"] = status
     _render_monitor_box(task)
     if status == "completed":
         output_path = _task_output_path(task)
@@ -685,6 +742,11 @@ def _render_task_status_once() -> None:
         if detail:
             st.warning(detail)
 
+    # Only boundary transitions need one full app rerun so static controls pick
+    # up resume/terminal state. Running progress never reruns the whole page.
+    if previous_status and status != previous_status and status in {"resumable", "completed", "failed", "cancelled"}:
+        st.rerun()
+
 
 if getattr(st, "fragment", None) is not None:
     @st.fragment(run_every=5)

config.py

@@ -137,6 +137,14 @@ def load_config():
             30,
             int(os.environ.get("SMARTPAPER_EXECUTOR_ACTIVITY_GRACE_SECONDS", "45")),
         ),
+        "executor_bootstrap_timeout_seconds": max(
+            5,
+            int(os.environ.get("SMARTPAPER_EXECUTOR_BOOTSTRAP_TIMEOUT_SECONDS", "20")),
+        ),
+        "executor_bootstrap_poll_ms": max(
+            50,
+            int(os.environ.get("SMARTPAPER_EXECUTOR_BOOTSTRAP_POLL_MS", "100")),
+        ),
         "manuscript_quality": {
             "abstract_max_words": int(os.environ.get("ABSTRACT_MAX_WORDS", "200")),
             "abstract_max_numeric_mentions": int(os.environ.get("ABSTRACT_MAX_NUMERIC_MENTIONS", "6")),

core/checkpoint_store.py

@@ -0,0 +1,71 @@
+# -*- coding: utf-8 -*-
+"""Lightweight task-scoped checkpoint persistence for SmartPaper executors.
+
+This module intentionally depends only on the Python standard library. It can
+be used immediately after a Worker claim without importing DeepSeek, RAG,
+statistics, document, or Streamlit modules.
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+def _atomic_json(path: Path, value: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
+    os.replace(temporary, path)
+
+
+class TaskCheckpointStore:
+    def __init__(self, task_dir: Path, request_hash: str, execution_epoch: int):
+        self.task_dir = Path(task_dir)
+        self.request_hash = str(request_hash)
+        self.execution_epoch = int(execution_epoch)
+        self.directory = self.task_dir / "checkpoints"
+        self.directory.mkdir(parents=True, exist_ok=True)
+        self.sequence = 0
+
+    def load_latest(self) -> dict[str, Any] | None:
+        candidates: list[tuple[int, Path, dict[str, Any]]] = []
+        for path in self.directory.glob("checkpoint_*.json"):
+            try:
+                payload = json.loads(path.read_text(encoding="utf-8"))
+                if not isinstance(payload, dict):
+                    continue
+                if payload.get("request_hash") != self.request_hash:
+                    continue
+                seq = int(payload.get("checkpoint_seq") or 0)
+                if seq <= 0:
+                    continue
+                candidates.append((seq, path, payload))
+            except Exception:
+                continue
+        if not candidates:
+            return None
+        seq, _, payload = max(candidates, key=lambda item: item[0])
+        self.sequence = seq
+        return payload
+
+    def save(self, stage: str, state: dict[str, Any]) -> tuple[int, str]:
+        self.sequence += 1
+        stage_name = re.sub(r"[^A-Za-z0-9_.-]", "_", str(stage or "checkpoint"))[:80]
+        payload = dict(state or {})
+        payload.update(
+            {
+                "request_hash": self.request_hash,
+                "checkpoint_seq": self.sequence,
+                "checkpoint_stage": stage_name,
+                "checkpoint_execution_epoch": self.execution_epoch,
+                "checkpoint_saved_at": datetime.now(timezone.utc).isoformat(),
+            }
+        )
+        path = self.directory / f"checkpoint_{self.sequence:05d}_{stage_name}.json"
+        _atomic_json(path, payload)
+        _atomic_json(self.directory / "latest.json", payload)
+        return self.sequence, stage_name

core/desktop.ini

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

core/generation_runner.py

@@ -18,6 +18,7 @@ from pathlib import Path
 from typing import Callable, Any
 
 from config import APP_CONFIG, AMJ_PROMPTS, ASQ_PROMPTS
+from core.checkpoint_store import TaskCheckpointStore
 from core.meta_extractor import extract_golden_metadata
 from core.pipeline_agent import SSciPipelineAgent
 from core.production_guard import build_validated_package, replace_abstract
@@ -29,13 +30,6 @@ from utils.file_parser import extract_text_from_files
 ProgressCallback = Callable[..., None]
 
 
-def _atomic_json(path: Path, value: dict[str, Any]) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    temporary = path.with_suffix(path.suffix + ".tmp")
-    temporary.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
-    os.replace(temporary, path)
-
-
 def _clean_trailing_references(text, keywords=("reference", "references", "reference list", "参考文献")):
     lines = str(text or "").split("\n")
     for i in range(len(lines) - 1, max(-1, len(lines) - 100), -1):
@@ -55,56 +49,6 @@ def _canonical_request_hash(request_data: dict[str, Any]) -> str:
     return hashlib.sha256(encoded).hexdigest()
 
 
-class TaskCheckpointStore:
-    def __init__(self, task_dir: Path, request_hash: str, execution_epoch: int):
-        self.task_dir = Path(task_dir)
-        self.request_hash = str(request_hash)
-        self.execution_epoch = int(execution_epoch)
-        self.directory = self.task_dir / "checkpoints"
-        self.directory.mkdir(parents=True, exist_ok=True)
-        self.sequence = 0
-
-    def load_latest(self) -> dict[str, Any] | None:
-        candidates: list[tuple[int, Path, dict[str, Any]]] = []
-        for path in self.directory.glob("checkpoint_*.json"):
-            try:
-                payload = json.loads(path.read_text(encoding="utf-8"))
-                if not isinstance(payload, dict):
-                    continue
-                if payload.get("request_hash") != self.request_hash:
-                    continue
-                seq = int(payload.get("checkpoint_seq") or 0)
-                if seq <= 0:
-                    continue
-                candidates.append((seq, path, payload))
-            except Exception:
-                continue
-        if not candidates:
-            return None
-        seq, _, payload = max(candidates, key=lambda item: item[0])
-        self.sequence = seq
-        return payload
-
-    def save(self, stage: str, state: dict[str, Any]) -> tuple[int, str]:
-        self.sequence += 1
-        stage_name = re.sub(r"[^A-Za-z0-9_.-]", "_", str(stage or "checkpoint"))[:80]
-        payload = dict(state or {})
-        payload.update(
-            {
-                "request_hash": self.request_hash,
-                "checkpoint_seq": self.sequence,
-                "checkpoint_stage": stage_name,
-                "checkpoint_execution_epoch": self.execution_epoch,
-                "checkpoint_saved_at": datetime.now(timezone.utc).isoformat(),
-            }
-        )
-        path = self.directory / f"checkpoint_{self.sequence:05d}_{stage_name}.json"
-        _atomic_json(path, payload)
-        latest = self.directory / "latest.json"
-        _atomic_json(latest, payload)
-        return self.sequence, stage_name
-
-
 class GenerationRunner:
     def __init__(
         self,

desktop.ini

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

managed_auth.py

@@ -29,34 +29,6 @@ from smartpaper_restore_bridge import edge_bridge
 EDGE_ASSERTION_COOKIE_NAME = "smartpaper_edge_assertion"
 EDGE_ASSERTION_AUDIENCE = "smartpaper-streamlit"
 ASSERTION_CLOCK_SKEW_SECONDS = 30
-ACTIVE_GENERATION_STATUSES = frozenset({"created", "running", "resumable", "cancelling"})
-KNOWN_GENERATION_STATUSES = ACTIVE_GENERATION_STATUSES | frozenset({"completed", "failed", "cancelled"})
-
-
-def _select_generation_task_for_session(
-    active_task: Any, latest_task: Any, tracked_task_id: str
-) -> dict[str, Any] | None:
-    """Choose only a genuinely current task for the Streamlit session.
-
-    ``latest_task`` is historical data.  It may update the terminal state of a
-    task that this Streamlit session already tracked, but it must never create a
-    current-task UI state on a fresh login/session by itself.
-    """
-    if isinstance(active_task, dict) and active_task.get("task_id"):
-        status = str(active_task.get("status") or "").strip().lower()
-        if status in ACTIVE_GENERATION_STATUSES:
-            return dict(active_task)
-
-    tracked = str(tracked_task_id or "").strip()
-    if not tracked or not isinstance(latest_task, dict):
-        return None
-    latest_id = str(latest_task.get("task_id") or "").strip()
-    latest_status = str(latest_task.get("status") or "").strip().lower()
-    if latest_id == tracked and latest_status in KNOWN_GENERATION_STATUSES:
-        return dict(latest_task)
-    return None
-
-
 def _runtime_value(name: str) -> str:
     value = os.environ.get(name, "")
     if not value:
@@ -158,6 +130,7 @@ def _verify_edge_assertion(token: str) -> dict[str, Any]:
 def _clear_generation_session_state() -> None:
     for key in (
         "sp_generation_task", "sp_tracked_task_id", "sp_current_node",
+        "sp_task_recovery_initialized", "sp_last_ui_task_status",
         "sp_generation_idempotency_key", "zip_buffer", "zip_filename",
         "is_running", "start_pipeline", "resume_pipeline", "run_mode",
         "rag_engine", "up_amj_ref", "up_amj_data", "up_asq_ref",
@@ -267,16 +240,9 @@ def get_account_snapshot(*, block: bool = False) -> dict[str, Any] | None:
     user = result.get("user")
     if isinstance(user, dict):
         st.session_state["sp_user"] = dict(user)
-    task = result.get("task")
-    latest_task = result.get("latest_task")
-    selected_task = _select_generation_task_for_session(
-        task, latest_task, str(st.session_state.get("sp_tracked_task_id") or "")
-    )
-    if selected_task is not None:
-        st.session_state["sp_generation_task"] = selected_task
-        st.session_state["sp_tracked_task_id"] = str(selected_task.get("task_id") or "")
-    else:
-        st.session_state["sp_generation_task"] = None
+    # R3.1 single-writer rule: account refresh owns identity/quota only.
+    # Task UI state is initialized by recover_generation_task() and thereafter
+    # updated only by generation_task_status()/explicit task actions.
     return result
 
 
@@ -307,12 +273,28 @@ def start_generation_task(*, idempotency_key: str, mode: str, interface_language
 
 
 def recover_generation_task() -> dict[str, Any]:
+    """Initialize current task state once from the authoritative Worker.
+
+    Account polling is intentionally forbidden from writing task state. This
+    recovery call is the only fresh-session initializer; later changes come
+    from generation_task_status() or explicit start/cancel operations.
+    """
     command_id = f"recover:{int(time.time() // 10)}"
     result = _edge_call("task_recover", command_id, {}, key_prefix="sp_task_recover")
     if result is None:
         st.stop()
     if not result.get("ok"):
         _raise_edge_error(result, "task_recover_failed")
+    _merge_quota_into_session(result)
+    task = result.get("task")
+    if isinstance(task, dict) and task.get("task_id"):
+        st.session_state["sp_generation_task"] = dict(task)
+        st.session_state["sp_tracked_task_id"] = str(task["task_id"])
+        st.session_state["sp_last_ui_task_status"] = str(task.get("status") or "")
+    else:
+        st.session_state["sp_generation_task"] = None
+        st.session_state["sp_last_ui_task_status"] = ""
+        st.session_state.pop("sp_tracked_task_id", None)
     return dict(result)
 
 

smartpaper_restore_bridge/desktop.ini

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

smartpaper_restore_bridge/frontend/desktop.ini

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

task_executor.py

@@ -1,10 +1,12 @@
 # -*- coding: utf-8 -*-
-"""SmartPaper R3-Free independent task executor.
-
-Launched by Streamlit as a detached subprocess. It claims a D1 task with a
-server-to-server service key, runs the headless generation pipeline, renews the
-lease only while real execution activity is recent, and atomically asks the
-Worker to commit/release quota at terminal completion.
+"""SmartPaper R3.1 independent task executor.
+
+Bootstrap is intentionally lightweight: the child process validates the local
+request, claims the D1 task, writes a local claim handshake, and only then
+imports the heavy paper-generation modules.  A Streamlit rerun/browser event
+therefore cannot be mistaken for successful executor startup, and import or
+pipeline failures after claim can be finalized through the authoritative
+Worker so quota is never left reserved by an unowned task.
 """
 from __future__ import annotations
 
@@ -16,24 +18,32 @@ import re
 import sys
 import threading
 import time
+from datetime import datetime, timezone
 from pathlib import Path
 from urllib.parse import urlsplit
 
 import requests
 
 from config import APP_CONFIG
-from core.generation_runner import GenerationRunner
-from core.pipeline_agent import (
-    DeepSeekAPIError, DeepSeekAuthenticationError, DeepSeekBalanceError, DeepSeekRequestError,
-)
 
+_log_handlers = [logging.StreamHandler(sys.stdout)]
+_executor_log_path = str(os.environ.get("SMARTPAPER_EXECUTOR_LOG_PATH") or "").strip()
+if _executor_log_path:
+    try:
+        Path(_executor_log_path).parent.mkdir(parents=True, exist_ok=True)
+        _log_handlers.append(logging.FileHandler(_executor_log_path, encoding="utf-8"))
+    except Exception:
+        pass
 logging.basicConfig(
     level=logging.INFO,
     format="%(asctime)s [%(levelname)s] %(message)s",
-    handlers=[logging.StreamHandler(sys.stdout)],
+    handlers=_log_handlers,
 )
 
 
+CLAIM_HANDSHAKE_FILENAME = "executor_claim.json"
+
+
 def _safe(value, limit=500):
     text = str(value or "")
     text = re.sub(r"(?i)(api[_ -]?key|authorization|bearer|token)\s*[:=]\s*[^\s,;]+", r"\1=[redacted]", text)
@@ -55,6 +65,17 @@ def _service_key() -> str:
     return value
 
 
+def _atomic_json(path: Path, value: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
+    os.replace(temporary, path)
+
+
+def _claim_marker_path(task_dir: Path) -> Path:
+    return task_dir / CLAIM_HANDSHAKE_FILENAME
+
+
 class ExecutorApiError(RuntimeError):
     def __init__(self, error: str, status_code: int = 0, payload: dict | None = None):
         super().__init__(str(error or "executor_api_error"))
@@ -64,19 +85,20 @@ class ExecutorApiError(RuntimeError):
 
 
 def _is_recoverable_generation_failure(exc: Exception) -> bool:
-    """Return True only for transient execution failures that can use checkpoints.
+    """Classify only post-claim execution failures.
 
-    Authentication/balance/request-validation failures are terminal and release
-    quota immediately. Transport/stream failures can be resumed from the last
-    durable task checkpoint without consuming a second quota unit.
+    This deliberately avoids importing the generation pipeline at module import
+    time.  DeepSeek exception class names are stable public names inside this
+    project and are inspected only after claim/bootstrap has succeeded.
     """
     if isinstance(exc, ExecutorApiError):
         return exc.status_code == 429 or exc.status_code >= 500
-    if isinstance(exc, (DeepSeekAuthenticationError, DeepSeekBalanceError, DeepSeekRequestError)):
+    class_name = exc.__class__.__name__
+    if class_name in {"DeepSeekAuthenticationError", "DeepSeekBalanceError", "DeepSeekRequestError"}:
         return False
     if isinstance(exc, (requests.exceptions.Timeout, requests.exceptions.ConnectionError)):
         return True
-    if isinstance(exc, DeepSeekAPIError):
+    if class_name == "DeepSeekAPIError":
         return True
     text = str(exc or "").lower()
     transient_markers = (
@@ -94,7 +116,7 @@ class ExecutorApi:
             "Authorization": f"Bearer {_service_key()}",
             "Accept": "application/json",
             "Content-Type": "application/json",
-            "User-Agent": "SmartPaper-R3-Free-Executor/1.0",
+            "User-Agent": "SmartPaper-R3.1-Executor/1.0",
         }
 
     def post(self, path: str, payload: dict, *, timeout=20) -> dict:
@@ -204,12 +226,11 @@ class ExecutionState:
     def _heartbeat_loop(self):
         interval = int(APP_CONFIG.get("executor_heartbeat_interval_seconds", 15))
         grace = int(APP_CONFIG.get("executor_activity_grace_seconds", 45))
-        # Claim itself already set the first heartbeat; delay before the next one.
         while not self._stop.wait(interval):
             payload, idle_seconds = self._payload()
             if idle_seconds > grace:
                 logging.warning(
-                    "[SP_EXECUTOR] heartbeat paused task_id=%s idle_s=%.1f node=%s",
+                    "[SP_EXECUTOR] heartbeat_paused task_id=%s idle_s=%.1f node=%s",
                     self.task_id, idle_seconds, payload.get("current_node"),
                 )
                 continue
@@ -231,17 +252,9 @@ class ExecutionState:
                         self.task_id, _safe(exc.error, 180),
                     )
                     return
-                logging.warning(
-                    "[SP_EXECUTOR] heartbeat_error task_id=%s error=%s",
-                    self.task_id, _safe(exc, 180),
-                )
+                logging.warning("[SP_EXECUTOR] heartbeat_error task_id=%s error=%s", self.task_id, _safe(exc, 180))
             except Exception as exc:
-                # A transport outage alone does not authorize another local
-                # writer. D1 lease/epoch ownership remains authoritative.
-                logging.warning(
-                    "[SP_EXECUTOR] heartbeat_error task_id=%s error=%s",
-                    self.task_id, _safe(exc, 180),
-                )
+                logging.warning("[SP_EXECUTOR] heartbeat_error task_id=%s error=%s", self.task_id, _safe(exc, 180))
 
 
 def _load_request(task_dir: Path) -> dict:
@@ -254,6 +267,70 @@ def _load_request(task_dir: Path) -> dict:
     return value
 
 
+def _bootstrap_claim(*, task_dir: Path, task_id: str, user_id: int, request_data: dict, api: ExecutorApi) -> tuple[dict, int, str]:
+    """Claim first and publish a PID-bound local handshake before heavy imports."""
+    request_hash = str(request_data.get("request_hash") or "").strip()
+    if not request_hash:
+        raise RuntimeError("request_hash_missing")
+    claim = api.claim(task_id=task_id, user_id=user_id, request_hash=request_hash)
+    task = dict(claim.get("task") or {})
+    execution = dict(claim.get("execution") or {})
+    epoch = int(execution.get("execution_epoch") or task.get("execution_epoch") or 0)
+    instance_id = str(execution.get("executor_instance_id") or "")
+    if epoch <= 0 or not instance_id:
+        raise RuntimeError("executor_claim_identity_missing")
+    if int(task.get("target_words") or 0) != int(request_data.get("target_words") or 0):
+        raise RuntimeError("target_words_claim_mismatch")
+    try:
+        _atomic_json(
+            _claim_marker_path(task_dir),
+            {
+                "task_id": task_id,
+                "user_id": int(user_id),
+                "request_hash": request_hash,
+                "pid": os.getpid(),
+                "execution_epoch": epoch,
+                "executor_instance_id": instance_id,
+                "claimed_at": datetime.now(timezone.utc).isoformat(),
+            },
+        )
+    except Exception as exc:
+        try:
+            api.finish(
+                {
+                    "task_id": task_id,
+                    "user_id": int(user_id),
+                    "execution_epoch": epoch,
+                    "executor_instance_id": instance_id,
+                    "status": "failed",
+                    "safe_error_code": "E_EXECUTOR_HANDSHAKE_FAILED",
+                    "safe_error_detail": _safe(exc, 800),
+                    "current_node": "executor_handshake_failed",
+                }
+            )
+        finally:
+            raise
+    return task, epoch, instance_id
+
+
+def _finish_bootstrap_failure(api: ExecutorApi, *, task_id: str, user_id: int, epoch: int, instance_id: str, exc: Exception) -> None:
+    try:
+        api.finish(
+            {
+                "task_id": task_id,
+                "user_id": user_id,
+                "execution_epoch": epoch,
+                "executor_instance_id": instance_id,
+                "status": "failed",
+                "safe_error_code": "E_EXECUTOR_BOOTSTRAP_FAILED",
+                "safe_error_detail": _safe(exc, 800),
+                "current_node": "executor_bootstrap_failed",
+            }
+        )
+    except Exception as finish_exc:
+        logging.warning("[SP_EXECUTOR] bootstrap_finalize_rejected task_id=%s error=%s", task_id, _safe(finish_exc, 220))
+
+
 def main() -> int:
     parser = argparse.ArgumentParser()
     parser.add_argument("--task-dir", required=True)
@@ -267,36 +344,19 @@ def main() -> int:
         raise RuntimeError("request_task_id_mismatch")
     if int(request_data.get("user_id") or 0) != int(args.user_id):
         raise RuntimeError("request_user_id_mismatch")
-    request_hash = str(request_data.get("request_hash") or "").strip()
-    if not request_hash:
-        raise RuntimeError("request_hash_missing")
 
     api = ExecutorApi()
-    claim = api.claim(task_id=args.task_id, user_id=args.user_id, request_hash=request_hash)
-    task = dict(claim.get("task") or {})
-    execution = dict(claim.get("execution") or {})
-    epoch = int(execution.get("execution_epoch") or task.get("execution_epoch") or 0)
-    instance_id = str(execution.get("executor_instance_id") or "")
-    if epoch <= 0 or not instance_id:
-        raise RuntimeError("executor_claim_identity_missing")
-    if int(task.get("target_words") or 0) != int(request_data.get("target_words") or 0):
-        raise RuntimeError("target_words_claim_mismatch")
-
-    state = ExecutionState(
+    task, epoch, instance_id = _bootstrap_claim(
+        task_dir=task_dir,
         task_id=args.task_id,
         user_id=args.user_id,
-        epoch=epoch,
-        instance_id=instance_id,
+        request_data=request_data,
         api=api,
     )
+    state = ExecutionState(task_id=args.task_id, user_id=args.user_id, epoch=epoch, instance_id=instance_id, api=api)
     state.start_heartbeat()
-    logging.info(
-        "[SP_EXECUTOR] claimed task_id=%s epoch=%s instance=%s",
-        args.task_id, epoch, instance_id[:12],
-    )
+    logging.info("[SP_EXECUTOR] claimed task_id=%s epoch=%s instance=%s", args.task_id, epoch, instance_id[:12])
 
-    # If D1 says a prior checkpoint exists but the free Space's local disk no
-    # longer has it (container rebuild), do not pretend resume is possible.
     remote_checkpoint_seq = int(task.get("checkpoint_seq") or 0)
     local_checkpoints = list((task_dir / "checkpoints").glob("checkpoint_*.json")) if (task_dir / "checkpoints").is_dir() else []
     if remote_checkpoint_seq > 0 and not local_checkpoints:
@@ -318,6 +378,11 @@ def main() -> int:
         return 2
 
     try:
+        # Heavy paper modules are deliberately imported only after D1 claim and
+        # the local claim handshake. Any import/init failure is now observable
+        # and can atomically release the reserved quota.
+        from core.generation_runner import GenerationRunner
+
         runner = GenerationRunner(
             task_dir=task_dir,
             request_data=request_data,
@@ -327,8 +392,6 @@ def main() -> int:
         )
         result = runner.run()
         state.touch(progress_pct=100, current_node="package_validated")
-        # One last accepted heartbeat publishes the final checkpoint metadata
-        # before the atomic terminal transaction.
         try:
             state.heartbeat_once()
         except Exception as exc:
@@ -350,19 +413,14 @@ def main() -> int:
         return 0
     except Exception as exc:
         if state.cancelled():
-            logging.warning(
-                "[SP_EXECUTOR] stopped_non_authoritative task_id=%s epoch=%s",
-                args.task_id, epoch,
-            )
+            logging.warning("[SP_EXECUTOR] stopped_non_authoritative task_id=%s epoch=%s", args.task_id, epoch)
             return 3
+        if isinstance(exc, (ImportError, ModuleNotFoundError)):
+            logging.exception("[SP_EXECUTOR] bootstrap_import_failed task_id=%s epoch=%s error=%s", args.task_id, epoch, _safe(exc, 240))
+            _finish_bootstrap_failure(api, task_id=args.task_id, user_id=args.user_id, epoch=epoch, instance_id=instance_id, exc=exc)
+            return 4
         if _is_recoverable_generation_failure(exc):
-            # Do not terminally fail or release quota. Stopping the real-execution
-            # heartbeat lets the Worker move this epoch to `resumable`; the next
-            # epoch continues from the latest immutable local checkpoint.
-            logging.exception(
-                "[SP_EXECUTOR] recoverable_interruption task_id=%s epoch=%s error=%s",
-                args.task_id, epoch, _safe(exc, 240),
-            )
+            logging.exception("[SP_EXECUTOR] recoverable_interruption task_id=%s epoch=%s error=%s", args.task_id, epoch, _safe(exc, 240))
             return 75
         logging.exception("[SP_EXECUTOR] failed task_id=%s epoch=%s error=%s", args.task_id, epoch, _safe(exc, 240))
         try:
@@ -379,8 +437,6 @@ def main() -> int:
                 }
             )
         except Exception as finish_exc:
-            # A newer epoch or user cancellation may already own/finalize the
-            # task. Never override it from a stale executor.
             logging.warning("[SP_EXECUTOR] failure_finalize_rejected task_id=%s error=%s", args.task_id, _safe(finish_exc, 220))
         return 1
     finally:

utils/desktop.ini

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