Spaces:
Sleeping
Sleeping
| # -*- 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 | |
| from core.advanced_method_registry import METHOD_FAMILIES, method_capability | |
| _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 _generation_failure_family(exc: Exception) -> str: | |
| """Classify post-claim execution failures for the P0 delivery-first contract.""" | |
| if isinstance(exc, ExecutorApiError): | |
| return "TRANSPORT" if exc.status_code == 429 or exc.status_code >= 500 else "AUTH_POLICY_SECURITY" | |
| class_name = exc.__class__.__name__ | |
| text = str(exc or "").casefold() | |
| if class_name in { | |
| "DeepSeekAuthenticationError", "DeepSeekBalanceError", | |
| "ModelAuthenticationError", "ModelBalanceError", | |
| }: | |
| return "AUTH_POLICY_SECURITY" | |
| stable_policy_markers = ( | |
| "trusted_product_policy", "trusted_product_route_", "request_hash_mismatch", | |
| "policy_snapshot", "entitlement", "authentication", "balance is insufficient", | |
| "rejected the request format (400)", "not found (404)", "parameters are invalid (422)", | |
| "unknown or unconfigured deepseek model role", "no deepseek model configured", | |
| ) | |
| if any(marker in text for marker in stable_policy_markers): | |
| return "AUTH_POLICY_SECURITY" | |
| if isinstance(exc, (requests.exceptions.Timeout, requests.exceptions.ConnectionError)): | |
| return "TRANSPORT" | |
| if isinstance(exc, requests.exceptions.HTTPError): | |
| response = getattr(exc, "response", None) | |
| status = int(getattr(response, "status_code", 0) or 0) | |
| if status in {408, 409, 425, 429, 500, 502, 503, 504}: | |
| return "TRANSPORT" | |
| if class_name in {"ModelProviderUnavailableError", "DeepSeekAPIError"}: | |
| return "TRANSPORT" | |
| if class_name == "DeepSeekRequestError": | |
| content_validation_markers = ( | |
| "final abstract", "front matter", "translation", "citation", "hypothesis", | |
| "manuscript", "qualitative", "glossary", "marker mismatch", "semantic", | |
| "must not contain", "empty after validation", "structured validation", | |
| ) | |
| return "CONTENT_VALIDATION" if any(marker in text for marker in content_validation_markers) else "AUTH_POLICY_SECURITY" | |
| if class_name == "ModelRequestError": | |
| # Gateway request errors are trusted route/request-shape failures; provider | |
| # availability has its own recoverable exception class. | |
| return "AUTH_POLICY_SECURITY" | |
| if class_name == "ModelGatewayError": | |
| transient_markers = ("unavailable", "timeout", "timed out", "connection", "provider") | |
| return "TRANSPORT" if any(marker in text for marker in transient_markers) else "CONTENT_VALIDATION" | |
| transient_markers = ( | |
| "connection", "timed out", "timeout", "temporarily unavailable", | |
| "streaming failed after", "api failed after", "stream was interrupted", | |
| "insufficient eligible references", | |
| ) | |
| if any(marker in text for marker in transient_markers): | |
| return "TRANSPORT" | |
| return "TERMINAL_INTEGRITY" | |
| def _is_recoverable_generation_failure(exc: Exception) -> bool: | |
| """Return True only for failures that should resume instead of refund/fail. | |
| CONTENT_VALIDATION reaches this layer only after the runner's local repair/ | |
| fallback paths were unable to complete the current attempt. A new executor | |
| attempt is still preferable to destroying the paper because model output is | |
| nondeterministic; stable auth/policy/request-shape faults remain terminal. | |
| """ | |
| return _generation_failure_family(exc) in {"TRANSPORT", "CONTENT_VALIDATION"} | |
| 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 _validate_request_method_family(request_data: dict, task: dict) -> None: | |
| """Validate new U1-R2 method identity while preserving legacy requests.""" | |
| raw_family = str((request_data or {}).get("method_family") or "").strip().casefold() | |
| if not raw_family: | |
| return | |
| if raw_family not in METHOD_FAMILIES: | |
| raise RuntimeError("request_method_family_invalid") | |
| capability = method_capability(raw_family) | |
| expected_lane = str(capability.get("legacy_policy_lane") or "").strip().casefold() | |
| request_lane = str((request_data or {}).get("derived_policy_lane") or "").strip().casefold() | |
| task_lane = str((task or {}).get("mode") or "").strip().casefold() | |
| if expected_lane not in {"amj", "asq"} or request_lane != expected_lane or task_lane != expected_lane: | |
| raise RuntimeError("request_method_family_lane_mismatch") | |
| if str(capability.get("capability_status") or "").strip().casefold() != "production_supported": | |
| raise RuntimeError("request_method_family_capability_blocked") | |
| 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") | |
| _validate_request_method_family(request_data, task) | |
| 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 | |
| runner = None | |
| 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") | |
| output_filename = str(result.get("output_filename") or "")[:240] | |
| 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": output_filename, | |
| # 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], | |
| } | |
| ) | |
| # P0-R2: private HF Bucket storage is a best-effort recovery layer only. | |
| # The authoritative completed/quota transition above must succeed first; | |
| # any storage failure below is nonfatal and cannot alter task accounting. | |
| try: | |
| from hf_bucket_store import archive_completed_zip | |
| archive_result = archive_completed_zip( | |
| user_id=args.user_id, | |
| task_id=args.task_id, | |
| output_filename=output_filename, | |
| output_path=str(result.get("output_path") or ""), | |
| ) | |
| if archive_result.get("stored"): | |
| logging.info( | |
| "[SP_EXECUTOR] bucket_backup_ok task_id=%s bytes=%s idempotent=%s", | |
| args.task_id, archive_result.get("size_bytes"), bool(archive_result.get("idempotent")), | |
| ) | |
| except Exception as exc: | |
| logging.warning( | |
| "[SP_EXECUTOR] bucket_backup_nonfatal task_id=%s error=%s", | |
| args.task_id, type(exc).__name__, | |
| ) | |
| 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)) | |
| recovery = {} | |
| if runner is not None: | |
| try: | |
| recovery = dict(runner.build_recovery_artifact(exc) or {}) | |
| except Exception as recovery_exc: | |
| logging.warning( | |
| "[SP_EXECUTOR] recovery_package_failed task_id=%s error=%s", | |
| args.task_id, _safe(recovery_exc, 220), | |
| ) | |
| 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", | |
| "output_filename": str(recovery.get("output_filename") or "")[:240], | |
| } | |
| ) | |
| 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()) | |