A5-R3-R5J-R4-R2 dormant dual-tier runtime

huggingface/task_executor.py

@@ -0,0 +1,766 @@
+# -*- coding: utf-8 -*-
+"""SmartPaper R3.2.2 independent task executor.
+
+Bootstrap is intentionally lightweight: the child process validates the local
+request, claims the D1 task, writes a local claim handshake, and only then
+imports the heavy paper-generation modules.  A Streamlit rerun/browser event
+therefore cannot be mistaken for successful executor startup, and import or
+pipeline failures after claim can be finalized through the authoritative
+Worker so quota is never left reserved by an unowned task.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import logging
+import os
+import re
+import secrets
+import shutil
+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.checkpoint_store import TaskCheckpointStore, checkpoint_authority_lock
+
+_log_handlers = [logging.StreamHandler(sys.stdout)]
+_executor_log_path = str(os.environ.get("SMARTPAPER_EXECUTOR_LOG_PATH") or "").strip()
+if _executor_log_path:
+    try:
+        Path(_executor_log_path).parent.mkdir(parents=True, exist_ok=True)
+        _log_handlers.append(logging.FileHandler(_executor_log_path, encoding="utf-8"))
+    except Exception:
+        pass
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    handlers=_log_handlers,
+)
+
+
+CLAIM_HANDSHAKE_FILENAME = "executor_claim.json"
+EXECUTOR_API_DEFAULT = "https://smartpaper-auth-relay-pages.pages.dev"
+
+
+def _bounded_env_float(name: str, default: float, minimum: float, maximum: float) -> float:
+    try:
+        value = float(os.environ.get(name, str(default)))
+    except (TypeError, ValueError):
+        value = float(default)
+    return max(float(minimum), min(float(maximum), value))
+
+
+# The current Worker startup lease is 90 seconds. R3.2 deliberately keeps the
+# default executor claim window below that lease while allowing materially
+# slower HF->Cloudflare TLS handshakes than R3.1.1's 4-second timeout.
+EXECUTOR_CLAIM_WINDOW_SECONDS = _bounded_env_float(
+    "SMARTPAPER_EXECUTOR_CLAIM_WINDOW_SECONDS", 70.0, 20.0, 80.0
+)
+EXECUTOR_CLAIM_REQUEST_TIMEOUT_SECONDS = _bounded_env_float(
+    "SMARTPAPER_EXECUTOR_CLAIM_REQUEST_TIMEOUT_SECONDS", 18.0, 5.0, 30.0
+)
+EXECUTOR_CLAIM_BACKOFF_INITIAL_SECONDS = _bounded_env_float(
+    "SMARTPAPER_EXECUTOR_CLAIM_BACKOFF_INITIAL_SECONDS", 1.0, 0.2, 5.0
+)
+EXECUTOR_CLAIM_BACKOFF_MAX_SECONDS = _bounded_env_float(
+    "SMARTPAPER_EXECUTOR_CLAIM_BACKOFF_MAX_SECONDS", 4.0, 1.0, 10.0
+)
+
+
+def _new_executor_instance_id() -> str:
+    """Create one stable identity for this executor claim attempt sequence."""
+    return f"exec_{secrets.token_urlsafe(18)}"
+
+
+def _safe(value, limit=500):
+    text = str(value or "")
+    text = re.sub(r"(?i)(api[_ -]?key|authorization|bearer|token)\s*[:=]\s*[^\s,;]+", r"\1=[redacted]", text)
+    return text[:limit]
+
+
+def _api_origin() -> str:
+    # Executor traffic intentionally uses the Pages Service-Binding relay.
+    # Browser/auth traffic continues to use SMARTPAPER_AUTH_API elsewhere.
+    value = str(os.environ.get("SMARTPAPER_EXECUTOR_API") or EXECUTOR_API_DEFAULT).strip().rstrip("/")
+    parsed = urlsplit(value)
+    if parsed.scheme != "https" or not parsed.netloc or parsed.path not in ("", "/") or parsed.query or parsed.fragment:
+        raise RuntimeError("SMARTPAPER_EXECUTOR_API_invalid")
+    return f"https://{parsed.netloc}"
+
+
+def _service_key() -> str:
+    value = str(os.environ.get("SMARTPAPER_SERVICE_KEY") or "").strip()
+    if not value:
+        raise RuntimeError("SMARTPAPER_SERVICE_KEY_missing")
+    return value
+
+
+def _atomic_json(path: Path, value: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
+    os.replace(temporary, path)
+
+
+def _claim_marker_path(task_dir: Path) -> Path:
+    return task_dir / CLAIM_HANDSHAKE_FILENAME
+
+
+class ExecutorApiError(RuntimeError):
+    def __init__(self, error: str, status_code: int = 0, payload: dict | None = None):
+        super().__init__(str(error or "executor_api_error"))
+        self.error = str(error or "executor_api_error")
+        self.status_code = int(status_code or 0)
+        self.payload = dict(payload or {})
+
+
+def _is_local_checkpoint_authority_failure(exc: Exception) -> bool:
+    return isinstance(exc, RuntimeError) and str(exc or "").startswith("checkpoint_authority_")
+
+
+def _is_recoverable_generation_failure(exc: Exception) -> bool:
+    """Classify only post-claim execution failures.
+
+    This deliberately avoids importing the generation pipeline at module import
+    time. Provider/model exception class names are inspected only after
+    claim/bootstrap has succeeded, so future provider adapters do not require
+    executor imports or vendor-specific branching.
+    """
+    if isinstance(exc, ExecutorApiError):
+        return exc.status_code == 429 or exc.status_code >= 500
+    class_name = exc.__class__.__name__
+    if class_name in {
+        "DeepSeekAuthenticationError", "DeepSeekBalanceError", "DeepSeekRequestError",
+        "ModelAuthenticationError", "ModelBalanceError", "ModelRequestError",
+        "ModelProviderUnavailableError", "ModelGatewayError",
+    }:
+        return False
+    if isinstance(exc, (requests.exceptions.Timeout, requests.exceptions.ConnectionError)):
+        return True
+    if class_name == "DeepSeekAPIError":
+        return True
+    text = str(exc or "").lower()
+    transient_markers = (
+        "connection", "timed out", "timeout", "temporarily unavailable",
+        "streaming failed after", "api failed after", "stream was interrupted",
+        "insufficient eligible references",
+    )
+    return any(marker in text for marker in transient_markers)
+
+
+class ExecutorApi:
+    def __init__(self):
+        self.origin = _api_origin()
+        self.headers = {
+            "Authorization": f"Bearer {_service_key()}",
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+            "User-Agent": "SmartPaper-R3.2.2-Executor/1.0",
+        }
+        # Keep direct networking first for the relay endpoint, with the ordinary
+        # environment-aware session only as a transport fallback. Business/HTTP
+        # errors never trigger fallback.
+        self._direct_session = requests.Session()
+        self._direct_session.trust_env = False
+        self._environment_session = requests.Session()
+
+    def post(self, path: str, payload: dict, *, timeout=20) -> dict:
+        url = self.origin + path
+        response = None
+        last_transport_error = None
+        transports = (
+            ("direct", self._direct_session),
+            ("environment", self._environment_session),
+        )
+        for index, (transport_name, session) in enumerate(transports):
+            try:
+                response = session.post(
+                    url,
+                    headers=self.headers,
+                    json=payload,
+                    timeout=timeout,
+                )
+                break
+            except (
+                requests.exceptions.SSLError,
+                requests.exceptions.Timeout,
+                requests.exceptions.ConnectionError,
+            ) as exc:
+                last_transport_error = exc
+                if index == 0:
+                    logging.warning(
+                        "[SP_EXECUTOR] transport_fallback path=%s from=direct to=environment error=%s",
+                        path, type(exc).__name__,
+                    )
+                    continue
+                raise
+
+        if response is None:
+            if last_transport_error is not None:
+                raise last_transport_error
+            raise RuntimeError("executor_transport_unavailable")
+
+        try:
+            data = response.json()
+        except Exception as exc:
+            raise RuntimeError(f"executor_worker_non_json_http_{response.status_code}") from exc
+        if not response.ok or data.get("ok") is False:
+            error = str(data.get("error") or data.get("reason") or f"HTTP_{response.status_code}")
+            raise ExecutorApiError(error, response.status_code, data)
+        return data
+
+    def claim(self, *, task_id: str, user_id: int, request_hash: str, executor_instance_id: str) -> dict:
+        payload = {
+            "task_id": task_id,
+            "user_id": user_id,
+            "request_hash": request_hash,
+            "executor_instance_id": executor_instance_id,
+        }
+        deadline = time.monotonic() + EXECUTOR_CLAIM_WINDOW_SECONDS
+        attempt = 0
+        while True:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise RuntimeError("executor_claim_deadline_exhausted")
+            attempt += 1
+            request_timeout = max(1.0, min(EXECUTOR_CLAIM_REQUEST_TIMEOUT_SECONDS, remaining))
+            try:
+                return self.post(
+                    "/executor/task/claim",
+                    payload,
+                    timeout=request_timeout,
+                )
+            except (
+                requests.exceptions.SSLError,
+                requests.exceptions.Timeout,
+                requests.exceptions.ConnectionError,
+            ) as exc:
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    raise
+                delay = min(
+                    EXECUTOR_CLAIM_BACKOFF_MAX_SECONDS,
+                    EXECUTOR_CLAIM_BACKOFF_INITIAL_SECONDS * (2 ** max(0, attempt - 1)),
+                    max(0.0, remaining - 0.25),
+                )
+                logging.warning(
+                    "[SP_EXECUTOR] claim_transport_retry task_id=%s attempt=%s error=%s delay=%.1fs remaining=%.1fs",
+                    task_id, attempt, type(exc).__name__, delay, remaining,
+                )
+                if delay <= 0:
+                    raise
+                time.sleep(delay)
+
+    def heartbeat(self, payload: dict) -> dict:
+        return self.post("/executor/task/heartbeat", payload)
+
+    def finish(self, payload: dict) -> dict:
+        return self.post("/executor/task/finish", payload, timeout=30)
+
+
+class ExecutionState:
+    def __init__(self, *, task_id: str, user_id: int, epoch: int, instance_id: str, api: ExecutorApi):
+        self.task_id = task_id
+        self.user_id = int(user_id)
+        self.epoch = int(epoch)
+        self.instance_id = str(instance_id)
+        self.api = api
+        self._lock = threading.Lock()
+        self._last_activity = time.monotonic()
+        self._progress_pct = 0
+        self._current_node = "executor_claimed"
+        self._checkpoint_seq = 0
+        self._checkpoint_stage = ""
+        self._ds_calls = 0
+        self._s2_calls = 0
+        self._oa_calls = 0
+        self._cancel = False
+        self._cancel_reason = ""
+        self._stop = threading.Event()
+        self._thread = None
+
+    def touch(self, *, progress_pct=None, current_node=None, checkpoint_seq=None, checkpoint_stage=None, ds_calls=None, s2_calls=None, oa_calls=None, activity_only=False):
+        with self._lock:
+            self._last_activity = time.monotonic()
+            # Request counters are observational metrics, not task-liveness
+            # authority. They may advance on activity-only callbacks while
+            # progress/node/checkpoint ownership remains unchanged.
+            if ds_calls is not None:
+                self._ds_calls = max(self._ds_calls, int(ds_calls))
+            if s2_calls is not None:
+                self._s2_calls = max(self._s2_calls, int(s2_calls))
+            if oa_calls is not None:
+                self._oa_calls = max(self._oa_calls, int(oa_calls))
+            if not activity_only:
+                if progress_pct is not None:
+                    self._progress_pct = max(self._progress_pct, max(0, min(100, int(progress_pct))))
+                if current_node:
+                    self._current_node = str(current_node)[:240]
+                if checkpoint_seq is not None and int(checkpoint_seq) >= self._checkpoint_seq:
+                    self._checkpoint_seq = int(checkpoint_seq)
+                    self._checkpoint_stage = str(checkpoint_stage or self._checkpoint_stage)[:120]
+
+    def cancelled(self) -> bool:
+        with self._lock:
+            return bool(self._cancel)
+
+    def cancel_reason(self) -> str:
+        with self._lock:
+            return str(self._cancel_reason or "")
+
+    def _mark_cancelled(self, reason: str) -> None:
+        with self._lock:
+            self._cancel = True
+            if reason:
+                self._cancel_reason = str(reason)[:80]
+
+    def start_heartbeat(self):
+        self._thread = threading.Thread(target=self._heartbeat_loop, name="smartpaper-executor-heartbeat", daemon=True)
+        self._thread.start()
+
+    def stop_heartbeat(self):
+        self._stop.set()
+        if self._thread and self._thread.is_alive():
+            self._thread.join(timeout=2)
+
+    def _payload(self):
+        with self._lock:
+            return {
+                "task_id": self.task_id,
+                "user_id": self.user_id,
+                "execution_epoch": self.epoch,
+                "executor_instance_id": self.instance_id,
+                "current_node": self._current_node,
+                "progress_pct": self._progress_pct,
+                "checkpoint_seq": self._checkpoint_seq,
+                "checkpoint_stage": self._checkpoint_stage,
+                "ds_calls": self._ds_calls,
+                "s2_calls": self._s2_calls,
+                "oa_calls": self._oa_calls,
+            }, time.monotonic() - self._last_activity
+
+    def heartbeat_once(self):
+        payload, _ = self._payload()
+        result = self.api.heartbeat(payload)
+        if result.get("cancel_requested"):
+            self._mark_cancelled("user_cancel")
+        return result
+
+    def _heartbeat_loop(self):
+        interval = int(APP_CONFIG.get("executor_heartbeat_interval_seconds", 15))
+        grace = int(APP_CONFIG.get("executor_activity_grace_seconds", 45))
+        while not self._stop.wait(interval):
+            payload, idle_seconds = self._payload()
+            # Long synchronous model calls can legitimately produce no Python
+            # callbacks for more than one Worker lease window.  Model activity
+            # is diagnostic only; process/epoch ownership is the liveness
+            # authority.  Therefore heartbeat must continue while this executor
+            # is alive and authoritative.
+            if idle_seconds > grace:
+                logging.info(
+                    "[SP_EXECUTOR] heartbeat_during_long_call task_id=%s idle_s=%.1f node=%s",
+                    self.task_id, idle_seconds, payload.get("current_node"),
+                )
+            try:
+                result = self.api.heartbeat(payload)
+                if result.get("cancel_requested"):
+                    self._mark_cancelled("user_cancel")
+                    return
+            except ExecutorApiError as exc:
+                if exc.error == "task_cancel_requested":
+                    self._mark_cancelled("user_cancel")
+                    logging.warning(
+                        "[SP_EXECUTOR] task_cancel_requested task_id=%s", self.task_id
+                    )
+                    return
+                if exc.error in {
+                    "executor_superseded", "generation_lock_lost",
+                    "task_already_finalized",
+                } or exc.status_code in {404, 409, 410}:
+                    self._mark_cancelled("authority_lost")
+                    logging.warning(
+                        "[SP_EXECUTOR] executor_no_longer_authoritative task_id=%s error=%s",
+                        self.task_id, _safe(exc.error, 180),
+                    )
+                    return
+                logging.warning("[SP_EXECUTOR] heartbeat_error task_id=%s error=%s", self.task_id, _safe(exc, 180))
+            except Exception as exc:
+                logging.warning("[SP_EXECUTOR] heartbeat_error task_id=%s error=%s", self.task_id, _safe(exc, 180))
+
+
+def _cleanup_user_cancelled_task_artifacts(task_dir: Path, request_data: dict) -> None:
+    """Delete only local, user-cancelled stage artifacts. D1 audit state remains authoritative."""
+    try:
+        token = str((request_data or {}).get("preview_token") or "").strip()
+        if re.fullmatch(r"[A-Za-z0-9_-]{32,128}", token):
+            preview = Path(__file__).resolve().parent / "static" / "previews" / f"{token}.json"
+            preview.unlink(missing_ok=True)
+    except Exception as exc:
+        logging.warning("[SP_EXECUTOR] cancel_preview_cleanup_error task_id=%s error=%s", task_dir.name, _safe(exc, 160))
+    try:
+        shutil.rmtree(task_dir, ignore_errors=True)
+    except Exception as exc:
+        logging.warning("[SP_EXECUTOR] cancel_task_cleanup_error task_id=%s error=%s", task_dir.name, _safe(exc, 160))
+
+
+def _snapshot_nullable_version(value):
+    text = str(value or "").strip()
+    return text or None
+
+
+def _trusted_task_policy_snapshot_envelope(snapshot: dict, resolved: dict) -> dict:
+    return {
+        "task_id": str(snapshot.get("task_id") or ""),
+        "snapshot_schema_version": str(snapshot.get("snapshot_schema_version") or ""),
+        "product_tier": str(snapshot.get("product_tier") or ""),
+        "product_policy_version": str(snapshot.get("product_policy_version") or ""),
+        "word_count_policy_version": str(snapshot.get("word_count_policy_version") or ""),
+        "model_policy_version": str(snapshot.get("model_policy_version") or ""),
+        "cost_policy_version": str(snapshot.get("cost_policy_version") or ""),
+        "billing_policy_version": _snapshot_nullable_version(snapshot.get("billing_policy_version")),
+        "trial_policy_version": _snapshot_nullable_version(snapshot.get("trial_policy_version")),
+        "entitlement_class": _snapshot_nullable_version(snapshot.get("entitlement_class")),
+        "min_body_words": int(snapshot.get("min_body_words") or 0),
+        "max_body_words": int(snapshot.get("max_body_words") or 0),
+        "counting_contract_version": str(snapshot.get("counting_contract_version") or ""),
+        "resolved_policy": resolved,
+    }
+
+
+def _validate_trusted_task_policy_snapshot(claim: dict, task: dict, request_data: dict) -> dict | None:
+    raw = claim.get("task_policy_snapshot")
+    if raw is None:
+        return None
+    if not isinstance(raw, dict):
+        raise RuntimeError("task_policy_snapshot_invalid")
+    snapshot = dict(raw)
+    resolved = snapshot.get("resolved_policy")
+    if not isinstance(resolved, dict):
+        raise RuntimeError("task_policy_snapshot_resolved_policy_missing")
+
+    snapshot_task_id = str(snapshot.get("task_id") or "").strip()
+    claimed_task_id = str(task.get("task_id") or "").strip()
+    request_task_id = str(request_data.get("task_id") or "").strip()
+    if not snapshot_task_id or snapshot_task_id != claimed_task_id or snapshot_task_id != request_task_id:
+        raise RuntimeError("task_policy_snapshot_task_mismatch")
+
+    tier = str(snapshot.get("product_tier") or "").strip().upper()
+    entitlement_class = str(snapshot.get("entitlement_class") or "").strip().upper()
+    if tier not in {"STANDARD", "FLAGSHIP"}:
+        raise RuntimeError("task_policy_snapshot_product_tier_invalid")
+    trial = resolved.get("trial") if isinstance(resolved.get("trial"), dict) else None
+    billing = resolved.get("billing") if isinstance(resolved.get("billing"), dict) else None
+    allowed_entitlements = []
+    if trial and str(trial.get("entitlement_class") or "").strip():
+        allowed_entitlements.append(str(trial.get("entitlement_class") or "").strip().upper())
+    if billing and str(billing.get("entitlement_class") or "").strip():
+        paid_class = str(billing.get("entitlement_class") or "").strip().upper()
+        if paid_class not in allowed_entitlements:
+            allowed_entitlements.append(paid_class)
+    if not entitlement_class or entitlement_class not in allowed_entitlements:
+        raise RuntimeError("task_policy_snapshot_entitlement_invalid")
+    if any(not value.startswith(f"{tier}_") for value in allowed_entitlements):
+        raise RuntimeError("task_policy_snapshot_entitlement_mismatch")
+
+    if str(resolved.get("snapshot_schema_version") or "") != str(snapshot.get("snapshot_schema_version") or ""):
+        raise RuntimeError("task_policy_snapshot_schema_mismatch")
+    if str(resolved.get("product_tier") or "").strip().upper() != tier:
+        raise RuntimeError("task_policy_snapshot_product_tier_mismatch")
+    version_fields = (
+        "product_policy_version",
+        "word_count_policy_version",
+        "model_policy_version",
+        "cost_policy_version",
+    )
+    for field in version_fields:
+        if str(resolved.get(field) or "") != str(snapshot.get(field) or ""):
+            raise RuntimeError("task_policy_snapshot_policy_version_mismatch")
+    for field in ("billing_policy_version", "trial_policy_version"):
+        if _snapshot_nullable_version(resolved.get(field)) != _snapshot_nullable_version(snapshot.get(field)):
+            raise RuntimeError("task_policy_snapshot_policy_version_mismatch")
+
+    if str(resolved.get("manuscript_mode") or "").strip().lower() != str(task.get("mode") or "").strip().lower():
+        raise RuntimeError("task_policy_snapshot_mode_mismatch")
+    target_words = int(task.get("target_words") or 0)
+    if target_words != int(request_data.get("target_words") or 0):
+        raise RuntimeError("target_words_claim_mismatch")
+    minimum = int(snapshot.get("min_body_words") or 0)
+    maximum = int(snapshot.get("max_body_words") or 0)
+    if minimum <= 0 or maximum < minimum or not minimum <= target_words <= maximum:
+        raise RuntimeError("task_policy_snapshot_word_range_mismatch")
+    word_count = resolved.get("word_count") if isinstance(resolved.get("word_count"), dict) else {}
+    if (
+        int(word_count.get("min_body_words") or 0) != minimum
+        or int(word_count.get("max_body_words") or 0) != maximum
+        or str(word_count.get("counting_contract_version") or "") != str(snapshot.get("counting_contract_version") or "")
+    ):
+        raise RuntimeError("task_policy_snapshot_word_policy_mismatch")
+
+    observed = hashlib.sha256(
+        json.dumps(
+            _trusted_task_policy_snapshot_envelope(snapshot, resolved),
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+        ).encode("utf-8")
+    ).hexdigest()
+    expected = str(snapshot.get("snapshot_sha256") or "").strip().lower()
+    if not re.fullmatch(r"[a-f0-9]{64}", expected) or observed != expected:
+        raise RuntimeError("task_policy_snapshot_sha_mismatch")
+    return snapshot
+
+
+def _load_request(task_dir: Path) -> dict:
+    path = task_dir / "request.json"
+    if not path.is_file():
+        raise RuntimeError("request_manifest_missing")
+    value = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(value, dict):
+        raise RuntimeError("request_manifest_invalid")
+    return value
+
+
+def _bootstrap_claim(*, task_dir: Path, task_id: str, user_id: int, request_data: dict, api: ExecutorApi) -> tuple[dict, int, str]:
+    """Claim first and publish a PID-bound local handshake before heavy imports."""
+    request_hash = str(request_data.get("request_hash") or "").strip()
+    if not request_hash:
+        raise RuntimeError("request_hash_missing")
+    instance_id = _new_executor_instance_id()
+    claim = api.claim(
+        task_id=task_id,
+        user_id=user_id,
+        request_hash=request_hash,
+        executor_instance_id=instance_id,
+    )
+    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)
+    claimed_instance_id = str(execution.get("executor_instance_id") or "")
+    if epoch <= 0 or not claimed_instance_id:
+        raise RuntimeError("executor_claim_identity_missing")
+    if claimed_instance_id != instance_id:
+        raise RuntimeError("executor_claim_identity_mismatch")
+    if int(task.get("target_words") or 0) != int(request_data.get("target_words") or 0):
+        raise RuntimeError("target_words_claim_mismatch")
+    trusted_task_policy_snapshot = _validate_trusted_task_policy_snapshot(
+        claim, task, request_data
+    )
+    if trusted_task_policy_snapshot is not None:
+        task["_trusted_task_policy_snapshot"] = trusted_task_policy_snapshot
+    try:
+        with checkpoint_authority_lock(task_dir):
+            _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,
+                    "worker_checkpoint_seq": int(task.get("checkpoint_seq") or 0),
+                    "worker_checkpoint_stage": str(task.get("checkpoint_stage") or "")[:120],
+                    "task_policy_snapshot_sha256": str(
+                        (trusted_task_policy_snapshot or {}).get("snapshot_sha256") or ""
+                    ),
+                    "task_policy_product_tier": str(
+                        (trusted_task_policy_snapshot or {}).get("product_tier") or ""
+                    ),
+                    "claimed_at": datetime.now(timezone.utc).isoformat(),
+                },
+            )
+    except Exception as exc:
+        try:
+            api.finish(
+                {
+                    "task_id": task_id,
+                    "user_id": int(user_id),
+                    "execution_epoch": epoch,
+                    "executor_instance_id": instance_id,
+                    "status": "failed",
+                    "safe_error_code": "E_EXECUTOR_HANDSHAKE_FAILED",
+                    "safe_error_detail": _safe(exc, 800),
+                    "current_node": "executor_handshake_failed",
+                }
+            )
+        finally:
+            raise
+    return task, epoch, instance_id
+
+
+def _finish_bootstrap_failure(api: ExecutorApi, *, task_id: str, user_id: int, epoch: int, instance_id: str, exc: Exception) -> None:
+    try:
+        api.finish(
+            {
+                "task_id": task_id,
+                "user_id": user_id,
+                "execution_epoch": epoch,
+                "executor_instance_id": instance_id,
+                "status": "failed",
+                "safe_error_code": "E_EXECUTOR_BOOTSTRAP_FAILED",
+                "safe_error_detail": _safe(exc, 800),
+                "current_node": "executor_bootstrap_failed",
+            }
+        )
+    except Exception as finish_exc:
+        logging.warning("[SP_EXECUTOR] bootstrap_finalize_rejected task_id=%s error=%s", task_id, _safe(finish_exc, 220))
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--task-dir", required=True)
+    parser.add_argument("--task-id", required=True)
+    parser.add_argument("--user-id", required=True, type=int)
+    args = parser.parse_args()
+
+    task_dir = Path(args.task_dir).resolve()
+    request_data = _load_request(task_dir)
+    if str(request_data.get("task_id") or "") != args.task_id:
+        raise RuntimeError("request_task_id_mismatch")
+    if int(request_data.get("user_id") or 0) != int(args.user_id):
+        raise RuntimeError("request_user_id_mismatch")
+
+    api = ExecutorApi()
+    task, epoch, instance_id = _bootstrap_claim(
+        task_dir=task_dir,
+        task_id=args.task_id,
+        user_id=args.user_id,
+        request_data=request_data,
+        api=api,
+    )
+    remote_checkpoint_seq = int(task.get("checkpoint_seq") or 0)
+    remote_checkpoint_stage = str(task.get("checkpoint_stage") or "")[:120]
+    state = ExecutionState(task_id=args.task_id, user_id=args.user_id, epoch=epoch, instance_id=instance_id, api=api)
+    if remote_checkpoint_seq > 0:
+        state.touch(
+            checkpoint_seq=remote_checkpoint_seq,
+            checkpoint_stage=remote_checkpoint_stage,
+            current_node=remote_checkpoint_stage or "executor_claimed",
+        )
+    state.start_heartbeat()
+    logging.info("[SP_EXECUTOR] claimed task_id=%s epoch=%s instance=%s", args.task_id, epoch, instance_id[:12])
+
+    confirmed_store = TaskCheckpointStore(
+        task_dir,
+        str(request_data.get("request_hash") or ""),
+        epoch,
+        task_id=args.task_id,
+        executor_instance_id=instance_id,
+        confirmed_checkpoint_seq=remote_checkpoint_seq,
+        confirmed_checkpoint_stage=remote_checkpoint_stage,
+    )
+    confirmed_checkpoint = confirmed_store.load_confirmed()
+    if remote_checkpoint_seq > 0 and confirmed_checkpoint is None:
+        try:
+            api.finish(
+                {
+                    "task_id": args.task_id,
+                    "user_id": args.user_id,
+                    "execution_epoch": epoch,
+                    "executor_instance_id": instance_id,
+                    "status": "failed",
+                    "safe_error_code": "E_CHECKPOINT_MISSING",
+                    "safe_error_detail": "Worker-confirmed checkpoint metadata does not match available local checkpoint material; quota released.",
+                    "current_node": "checkpoint_missing",
+                }
+            )
+        finally:
+            state.stop_heartbeat()
+        return 2
+
+    try:
+        # Heavy paper modules are deliberately imported only after D1 claim and
+        # the local claim handshake. Any import/init failure is now observable
+        # and can atomically release the reserved quota.
+        from core.generation_runner import GenerationRunner
+
+        runner = GenerationRunner(
+            task_dir=task_dir,
+            request_data=request_data,
+            execution_epoch=epoch,
+            executor_instance_id=instance_id,
+            confirmed_checkpoint_seq=remote_checkpoint_seq,
+            confirmed_checkpoint_stage=remote_checkpoint_stage,
+            trusted_task_policy_snapshot=task.get("_trusted_task_policy_snapshot"),
+            progress_callback=state.touch,
+            cancel_checker=state.cancelled,
+        )
+        result = runner.run()
+        state.touch(progress_pct=100, current_node="package_validated")
+        try:
+            state.heartbeat_once()
+        except Exception as exc:
+            logging.warning("[SP_EXECUTOR] final_heartbeat_error task_id=%s error=%s", args.task_id, _safe(exc, 180))
+        if state.cancelled():
+            raise RuntimeError("task_cancel_requested")
+        api.finish(
+            {
+                "task_id": args.task_id,
+                "user_id": args.user_id,
+                "execution_epoch": epoch,
+                "executor_instance_id": instance_id,
+                "status": "completed",
+                "current_node": "package_validated",
+                "output_filename": str(result.get("output_filename") or "")[:240],
+                # R14-TQ2-PQ1-R3-R1: completed tasks may carry a nonfatal,
+                # backend-trusted quality warning. Worker/D1 already persists
+                # safe_error_code/detail for completed terminal transitions;
+                # quota semantics remain the ordinary completed/committed path.
+                "safe_error_code": str(result.get("safe_error_code") or "")[:80],
+                "safe_error_detail": str(result.get("safe_error_detail") or "")[:800],
+            }
+        )
+        logging.info("[SP_EXECUTOR] completed task_id=%s epoch=%s", args.task_id, epoch)
+        return 0
+    except Exception as exc:
+        if state.cancelled():
+            reason = state.cancel_reason()
+            if reason == "user_cancel":
+                _cleanup_user_cancelled_task_artifacts(task_dir, request_data)
+                logging.info("[SP_EXECUTOR] cancelled_and_cleaned task_id=%s epoch=%s", args.task_id, epoch)
+            else:
+                logging.warning("[SP_EXECUTOR] stopped_non_authoritative task_id=%s epoch=%s reason=%s", args.task_id, epoch, reason or "unknown")
+            return 3
+        if _is_local_checkpoint_authority_failure(exc):
+            logging.warning(
+                "[SP_EXECUTOR] stopped_non_authoritative task_id=%s epoch=%s reason=local_checkpoint_authority_lost error=%s",
+                args.task_id, epoch, _safe(exc, 180),
+            )
+            return 3
+        if isinstance(exc, (ImportError, ModuleNotFoundError)):
+            logging.exception("[SP_EXECUTOR] bootstrap_import_failed task_id=%s epoch=%s error=%s", args.task_id, epoch, _safe(exc, 240))
+            _finish_bootstrap_failure(api, task_id=args.task_id, user_id=args.user_id, epoch=epoch, instance_id=instance_id, exc=exc)
+            return 4
+        if _is_recoverable_generation_failure(exc):
+            logging.exception("[SP_EXECUTOR] recoverable_interruption task_id=%s epoch=%s error=%s", args.task_id, epoch, _safe(exc, 240))
+            return 75
+        logging.exception("[SP_EXECUTOR] failed task_id=%s epoch=%s error=%s", args.task_id, epoch, _safe(exc, 240))
+        try:
+            api.finish(
+                {
+                    "task_id": args.task_id,
+                    "user_id": args.user_id,
+                    "execution_epoch": epoch,
+                    "executor_instance_id": instance_id,
+                    "status": "failed",
+                    "safe_error_code": "E_GENERATION_FAILED",
+                    "safe_error_detail": _safe(exc, 800),
+                    "current_node": "generation_failed",
+                }
+            )
+        except Exception as finish_exc:
+            logging.warning("[SP_EXECUTOR] failure_finalize_rejected task_id=%s error=%s", args.task_id, _safe(finish_exc, 220))
+        return 1
+    finally:
+        state.stop_heartbeat()
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())