R14-RESUME-R1: worker-confirmed checkpoint and local epoch fencing

core/checkpoint_store.py

@@ -10,9 +10,20 @@ from __future__ import annotations
 import json
 import os
 import re
+from contextlib import contextmanager
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Any
+from typing import Any, Iterator
+
+try:  # Hugging Face production runs on Linux.
+    import fcntl  # type: ignore
+except ImportError:  # pragma: no cover - Windows-only local development fallback.
+    fcntl = None
+    import msvcrt  # type: ignore
+
+
+CLAIM_HANDSHAKE_FILENAME = "executor_claim.json"
+AUTHORITY_LOCK_FILENAME = ".checkpoint_authority.lock"
 
 
 def _atomic_json(path: Path, value: dict[str, Any]) -> None:
@@ -22,50 +33,194 @@ def _atomic_json(path: Path, value: dict[str, Any]) -> None:
     os.replace(temporary, path)
 
 
+@contextmanager
+def checkpoint_authority_lock(task_dir: Path) -> Iterator[None]:
+    """Serialize local claim-marker replacement and checkpoint writes.
+
+    Worker/D1 remains the global authority. This small process lock only closes
+    the local race where a superseded executor could otherwise validate an old
+    marker immediately before a newer epoch replaces it, then write afterwards.
+    """
+    lock_path = Path(task_dir) / AUTHORITY_LOCK_FILENAME
+    lock_path.parent.mkdir(parents=True, exist_ok=True)
+    handle = lock_path.open("a+b")
+    try:
+        if fcntl is not None:
+            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
+        else:  # pragma: no cover - Windows-only local development fallback.
+            handle.seek(0, os.SEEK_END)
+            if handle.tell() == 0:
+                handle.write(b"0")
+                handle.flush()
+            handle.seek(0)
+            msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1)
+        yield
+    finally:
+        try:
+            if fcntl is not None:
+                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
+            else:  # pragma: no cover - Windows-only local development fallback.
+                handle.seek(0)
+                msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
+        finally:
+            handle.close()
+
+
 class TaskCheckpointStore:
-    def __init__(self, task_dir: Path, request_hash: str, execution_epoch: int):
+    def __init__(
+        self,
+        task_dir: Path,
+        request_hash: str,
+        execution_epoch: int,
+        *,
+        task_id: str = "",
+        executor_instance_id: str = "",
+        confirmed_checkpoint_seq: int | None = None,
+        confirmed_checkpoint_stage: str = "",
+    ):
         self.task_dir = Path(task_dir)
         self.request_hash = str(request_hash)
         self.execution_epoch = int(execution_epoch)
+        self.task_id = str(task_id or "")
+        self.executor_instance_id = str(executor_instance_id or "")
+        if bool(self.task_id) != bool(self.executor_instance_id):
+            raise ValueError("checkpoint_authority_identity_incomplete")
+        self.confirmed_checkpoint_seq = (
+            None if confirmed_checkpoint_seq is None else max(0, int(confirmed_checkpoint_seq))
+        )
+        self.confirmed_checkpoint_stage = str(confirmed_checkpoint_stage or "")[:120]
         self.directory = self.task_dir / "checkpoints"
         self.directory.mkdir(parents=True, exist_ok=True)
-        self.sequence = 0
+        self.sequence = int(self.confirmed_checkpoint_seq or 0)
+
+    def _read_payload(self, path: Path) -> dict[str, Any] | None:
+        try:
+            payload = json.loads(path.read_text(encoding="utf-8"))
+        except Exception:
+            return None
+        if not isinstance(payload, dict):
+            return None
+        if payload.get("request_hash") != self.request_hash:
+            return None
+        try:
+            seq = int(payload.get("checkpoint_seq") or 0)
+            writer_epoch = int(payload.get("checkpoint_execution_epoch") or 0)
+        except (TypeError, ValueError):
+            return None
+        if seq <= 0 or writer_epoch <= 0 or writer_epoch > self.execution_epoch:
+            return None
+        return payload
+
+    def load_confirmed(
+        self,
+        checkpoint_seq: int | None = None,
+        checkpoint_stage: str | None = None,
+    ) -> dict[str, Any] | None:
+        """Load only the exact checkpoint sequence acknowledged by Worker/D1.
+
+        Unacknowledged local snapshots from an older network-partitioned epoch
+        are intentionally ignored even when they have a larger sequence number.
+        """
+        seq = self.confirmed_checkpoint_seq if checkpoint_seq is None else max(0, int(checkpoint_seq))
+        stage = self.confirmed_checkpoint_stage if checkpoint_stage is None else str(checkpoint_stage or "")[:120]
+        if seq is None:
+            raise ValueError("confirmed_checkpoint_seq_required")
+        self.sequence = int(seq)
+        if seq <= 0:
+            return None
+
+        candidates: list[dict[str, Any]] = []
+        for path in self.directory.glob(f"checkpoint_{seq:05d}_*.json"):
+            payload = self._read_payload(path)
+            if payload is None:
+                continue
+            if int(payload.get("checkpoint_seq") or 0) != seq:
+                continue
+            if stage and str(payload.get("checkpoint_stage") or "") != stage:
+                continue
+            candidates.append(payload)
+        if len(candidates) != 1:
+            return None
+        return candidates[0]
 
     def load_latest(self) -> dict[str, Any] | None:
+        # Authoritative production callers always pin recovery to the exact
+        # Worker-confirmed checkpoint. Keep the legacy scan only for isolated
+        # compatibility/tests that do not yet supply confirmed metadata.
+        if self.confirmed_checkpoint_seq is not None:
+            return self.load_confirmed()
+
         candidates: list[tuple[int, Path, dict[str, Any]]] = []
         for path in self.directory.glob("checkpoint_*.json"):
-            try:
-                payload = json.loads(path.read_text(encoding="utf-8"))
-                if not isinstance(payload, dict):
-                    continue
-                if payload.get("request_hash") != self.request_hash:
-                    continue
-                seq = int(payload.get("checkpoint_seq") or 0)
-                if seq <= 0:
-                    continue
-                candidates.append((seq, path, payload))
-            except Exception:
+            payload = self._read_payload(path)
+            if payload is None:
                 continue
+            seq = int(payload.get("checkpoint_seq") or 0)
+            candidates.append((seq, path, payload))
         if not candidates:
             return None
         seq, _, payload = max(candidates, key=lambda item: item[0])
         self.sequence = seq
         return payload
 
+    def _assert_local_authority_locked(self) -> None:
+        if not self.task_id:
+            return
+        marker_path = self.task_dir / CLAIM_HANDSHAKE_FILENAME
+        try:
+            marker = json.loads(marker_path.read_text(encoding="utf-8"))
+        except Exception as exc:
+            raise RuntimeError("checkpoint_authority_marker_unavailable") from exc
+        if not isinstance(marker, dict):
+            raise RuntimeError("checkpoint_authority_marker_invalid")
+
+        expected = {
+            "task_id": self.task_id,
+            "request_hash": self.request_hash,
+            "execution_epoch": self.execution_epoch,
+            "executor_instance_id": self.executor_instance_id,
+        }
+        for key, value in expected.items():
+            actual = marker.get(key)
+            if key == "execution_epoch":
+                try:
+                    actual = int(actual or 0)
+                except (TypeError, ValueError):
+                    actual = 0
+            else:
+                actual = str(actual or "")
+            if actual != value:
+                raise RuntimeError("checkpoint_authority_lost")
+
+        if self.confirmed_checkpoint_seq is not None:
+            try:
+                marker_base_seq = int(marker.get("worker_checkpoint_seq") or 0)
+            except (TypeError, ValueError):
+                marker_base_seq = -1
+            marker_base_stage = str(marker.get("worker_checkpoint_stage") or "")[:120]
+            if marker_base_seq != int(self.confirmed_checkpoint_seq):
+                raise RuntimeError("checkpoint_authority_base_mismatch")
+            if self.confirmed_checkpoint_stage and marker_base_stage != self.confirmed_checkpoint_stage:
+                raise RuntimeError("checkpoint_authority_base_mismatch")
+
     def save(self, stage: str, state: dict[str, Any]) -> tuple[int, str]:
-        self.sequence += 1
         stage_name = re.sub(r"[^A-Za-z0-9_.-]", "_", str(stage or "checkpoint"))[:80]
-        payload = dict(state or {})
-        payload.update(
-            {
-                "request_hash": self.request_hash,
-                "checkpoint_seq": self.sequence,
-                "checkpoint_stage": stage_name,
-                "checkpoint_execution_epoch": self.execution_epoch,
-                "checkpoint_saved_at": datetime.now(timezone.utc).isoformat(),
-            }
-        )
-        path = self.directory / f"checkpoint_{self.sequence:05d}_{stage_name}.json"
-        _atomic_json(path, payload)
-        _atomic_json(self.directory / "latest.json", payload)
+        with checkpoint_authority_lock(self.task_dir):
+            self._assert_local_authority_locked()
+            next_sequence = self.sequence + 1
+            payload = dict(state or {})
+            payload.update(
+                {
+                    "request_hash": self.request_hash,
+                    "checkpoint_seq": next_sequence,
+                    "checkpoint_stage": stage_name,
+                    "checkpoint_execution_epoch": self.execution_epoch,
+                    "checkpoint_executor_instance_id": self.executor_instance_id,
+                    "checkpoint_saved_at": datetime.now(timezone.utc).isoformat(),
+                }
+            )
+            path = self.directory / f"checkpoint_{next_sequence:05d}_{stage_name}.json"
+            _atomic_json(path, payload)
+            _atomic_json(self.directory / "latest.json", payload)
+            self.sequence = next_sequence
         return self.sequence, stage_name

core/generation_runner.py

@@ -146,12 +146,20 @@ class GenerationRunner:
         task_dir: str | os.PathLike,
         request_data: dict[str, Any],
         execution_epoch: int,
+        executor_instance_id: str = "",
+        confirmed_checkpoint_seq: int | None = None,
+        confirmed_checkpoint_stage: str = "",
         progress_callback: ProgressCallback | None = None,
         cancel_checker: Callable[[], bool] | None = None,
     ):
         self.task_dir = Path(task_dir)
         self.request = dict(request_data)
         self.execution_epoch = int(execution_epoch)
+        self.executor_instance_id = str(executor_instance_id or "")
+        self.confirmed_checkpoint_seq = (
+            None if confirmed_checkpoint_seq is None else max(0, int(confirmed_checkpoint_seq))
+        )
+        self.confirmed_checkpoint_stage = str(confirmed_checkpoint_stage or "")[:120]
         self.progress_callback = progress_callback or (lambda **_: None)
         self.cancel_checker = cancel_checker or (lambda: False)
         expected_hash = str(self.request.get("request_hash") or "")
@@ -159,7 +167,15 @@ class GenerationRunner:
         if not expected_hash or expected_hash != actual_hash:
             raise RuntimeError("request_hash_mismatch")
         self.request_hash = expected_hash
-        self.checkpoints = TaskCheckpointStore(self.task_dir, self.request_hash, self.execution_epoch)
+        self.checkpoints = TaskCheckpointStore(
+            self.task_dir,
+            self.request_hash,
+            self.execution_epoch,
+            task_id=str(self.request.get("task_id") or "") if self.executor_instance_id else "",
+            executor_instance_id=self.executor_instance_id,
+            confirmed_checkpoint_seq=self.confirmed_checkpoint_seq,
+            confirmed_checkpoint_stage=self.confirmed_checkpoint_stage,
+        )
         self.output_dir = self.task_dir / "outputs"
         self.output_dir.mkdir(parents=True, exist_ok=True)
         self.agent = None
@@ -1516,7 +1532,13 @@ class GenerationRunner:
         quant_engine = get_quantitative_engine()
         qual_engine = get_qualitative_engine(agent)
 
-        state = self.checkpoints.load_latest() or {}
+        if self.confirmed_checkpoint_seq is None:
+            # Compatibility path for isolated unit helpers that do not represent
+            # a claimed production executor. Production always supplies Worker
+            # checkpoint metadata through task_executor.
+            state = self.checkpoints.load_latest() or {}
+        else:
+            state = self.checkpoints.load_confirmed() or {}
         quality_diagnostics = list(state.get("quality_diagnostics") or [])
         if state:
             self._preview_english_text = str(state.get('full_draft_en') or '')

task_executor.py

@@ -27,6 +27,7 @@ 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()
@@ -119,6 +120,10 @@ class ExecutorApiError(RuntimeError):
         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.
 
@@ -432,18 +437,21 @@ def _bootstrap_claim(*, task_dir: Path, task_id: str, user_id: int, request_data
     if int(task.get("target_words") or 0) != int(request_data.get("target_words") or 0):
         raise RuntimeError("target_words_claim_mismatch")
     try:
-        _atomic_json(
-            _claim_marker_path(task_dir),
-            {
-                "task_id": task_id,
-                "user_id": int(user_id),
-                "request_hash": request_hash,
-                "pid": os.getpid(),
-                "execution_epoch": epoch,
-                "executor_instance_id": instance_id,
-                "claimed_at": datetime.now(timezone.utc).isoformat(),
-            },
-        )
+        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(
@@ -503,13 +511,29 @@ def main() -> int:
         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])
 
-    remote_checkpoint_seq = int(task.get("checkpoint_seq") or 0)
-    local_checkpoints = list((task_dir / "checkpoints").glob("checkpoint_*.json")) if (task_dir / "checkpoints").is_dir() else []
-    if remote_checkpoint_seq > 0 and not local_checkpoints:
+    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(
                 {
@@ -519,7 +543,7 @@ def main() -> int:
                     "executor_instance_id": instance_id,
                     "status": "failed",
                     "safe_error_code": "E_CHECKPOINT_MISSING",
-                    "safe_error_detail": "Checkpoint metadata exists but local checkpoint files are unavailable after runtime replacement; quota released.",
+                    "safe_error_detail": "Worker-confirmed checkpoint metadata does not match available local checkpoint material; quota released.",
                     "current_node": "checkpoint_missing",
                 }
             )
@@ -537,6 +561,9 @@ def main() -> int:
             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,
         )
@@ -570,6 +597,12 @@ def main() -> int:
             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)