Blame for task_executor.py

# -*- 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")
    if entitlement_class not in {"STANDARD_TRIAL", "STANDARD_PAID", "FLAGSHIP_PAID"}:
        raise RuntimeError("task_policy_snapshot_entitlement_invalid")
    if tier == "STANDARD" and entitlement_class == "FLAGSHIP_PAID":
        raise RuntimeError("task_policy_snapshot_entitlement_mismatch")
    if tier == "FLAGSHIP" and entitlement_class != "FLAGSHIP_PAID":
        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())