# -*- 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 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. DeepSeek exception class names are stable public names inside this project and are inspected only after claim/bootstrap has succeeded. """ if isinstance(exc, ExecutorApiError): return exc.status_code == 429 or exc.status_code >= 500 class_name = exc.__class__.__name__ if class_name in {"DeepSeekAuthenticationError", "DeepSeekBalanceError", "DeepSeekRequestError"}: return False if isinstance(exc, (requests.exceptions.Timeout, requests.exceptions.ConnectionError)): return True if class_name == "DeepSeekAPIError": return True text = str(exc or "").lower() transient_markers = ( "connection", "timed out", "timeout", "temporarily unavailable", "streaming failed after", "api failed after", "stream was interrupted", "insufficient eligible references", ) return any(marker in text for marker in transient_markers) class ExecutorApi: def __init__(self): self.origin = _api_origin() self.headers = { "Authorization": f"Bearer {_service_key()}", "Accept": "application/json", "Content-Type": "application/json", "User-Agent": "SmartPaper-R3.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 _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") 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], "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, progress_callback=state.touch, cancel_checker=state.cancelled, ) result = runner.run() state.touch(progress_pct=100, current_node="package_validated") try: state.heartbeat_once() except Exception as exc: logging.warning("[SP_EXECUTOR] final_heartbeat_error task_id=%s error=%s", args.task_id, _safe(exc, 180)) if state.cancelled(): raise RuntimeError("task_cancel_requested") api.finish( { "task_id": args.task_id, "user_id": args.user_id, "execution_epoch": epoch, "executor_instance_id": instance_id, "status": "completed", "current_node": "package_validated", "output_filename": str(result.get("output_filename") or "")[:240], } ) logging.info("[SP_EXECUTOR] completed task_id=%s epoch=%s", args.task_id, epoch) return 0 except Exception as exc: if state.cancelled(): 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())