From 53960645ef65b507acfd7f70693dc8a788c1e033 Mon Sep 17 00:00:00 2001
From: ReallocAll <ReallocAll@outlook.com>
Date: Thu, 15 Oct 2026 10:27:16 +0800
Subject: [PATCH 1/4] fix: enforce worker ownership, lifecycle and durable RPC
 handoff

---
 src/endstone_endkeep/plugin.py          |   5 +-
 src/endstone_endkeep/worker/client.py   | 176 +++++++++++++++------
 src/endstone_endkeep/worker/process.py  |  55 +++++++
 src/endstone_endkeep/worker/protocol.py |   4 +
 src/endstone_endkeep/worker/server.py   | 195 +++++++++++++++++++++---
 5 files changed, 370 insertions(+), 65 deletions(-)
 create mode 100644 src/endstone_endkeep/worker/process.py

diff --git a/src/endstone_endkeep/plugin.py b/src/endstone_endkeep/plugin.py
index e7044169..d9a9f75f 100644
--- a/src/endstone_endkeep/plugin.py
+++ b/src/endstone_endkeep/plugin.py
@@ -264,7 +264,10 @@ def _close_runtime(self) -> None:
         if capture is not None:
             capture.close()
         if repository is not None:
-            repository.detach()
+            try:
+                repository.detach()
+            except WorkerError as exc:
+                self.logger.warning(f"Could not notify repository worker of control handoff: {exc}")
 
     def _tick(self) -> None:
         capture = self._capture
diff --git a/src/endstone_endkeep/worker/client.py b/src/endstone_endkeep/worker/client.py
index 8c62e03f..5d645976 100644
--- a/src/endstone_endkeep/worker/client.py
+++ b/src/endstone_endkeep/worker/client.py
@@ -10,6 +10,7 @@
 from pathlib import Path
 from typing import Any
 
+from .process import process_identity
 from .protocol import PROTOCOL_VERSION, RuntimeInfo, package_version, receive_message, send_message
 
 
@@ -42,52 +43,105 @@ def __init__(self, runtime_path: Path, runtime: RuntimeInfo, *, desired_priority
         self.runtime = runtime
         self.desired_priority = desired_priority
         self._status: dict[str, Any] = {"job": {"state": "idle"}}
+        self._controller_generation: int | None = None
 
     @classmethod
     def connect_or_start(cls, storage_root: Path, *, priority: str) -> RepositoryWorkerClient:
         storage_root.mkdir(parents=True, exist_ok=True)
         runtime_path = storage_root / "worker-runtime.json"
-
         existing = cls._load_runtime(runtime_path)
         if existing is not None:
-            if existing.protocol != PROTOCOL_VERSION:
-                if cls._pid_alive(existing.pid):
+            if Path(existing.storage_root).resolve() != storage_root.resolve():
+                raise WorkerError("repository worker runtime points at a different storage root")
+            if existing.protocol != PROTOCOL_VERSION and cls._runtime_alive(existing):
+                raise WorkerError(
+                    f"repository worker protocol mismatch: {existing.protocol} != {PROTOCOL_VERSION}; "
+                    "the previous worker must finish before starting another"
+                )
+            if existing.protocol == PROTOCOL_VERSION:
+                client = cls(runtime_path, existing, desired_priority=priority)
+                last_error: Exception | None = None
+                for attempt in range(4):
+                    try:
+                        response = client._rpc({"command": "ping"}, timeout=1.0)
+                    except WorkerError as exc:
+                        last_error = exc
+                        if not cls._runtime_alive(existing):
+                            break
+                        if attempt < 3:
+                            time.sleep(0.2)
+                        continue
+                    status = response["status"]
+                    worker = status.get("worker", {})
+                    if int(worker.get("pid", 0)) != existing.pid:
+                        raise WorkerError("repository runtime PID does not match the responding worker")
+                    if existing.process_start and process_identity(existing.pid) != existing.process_start:
+                        raise WorkerError("repository runtime process identity changed during reconnect")
+                    client._status = status
+                    if "controller_start" in worker:
+                        identity = process_identity(os.getpid())
+                        if identity is None:
+                            raise WorkerError("cannot identify the current controller process")
+                        try:
+                            response = client._rpc(
+                                {
+                                    "command": "attach",
+                                    "controller_pid": os.getpid(),
+                                    "controller_start": identity,
+                                    "controller_generation": worker.get("controller_generation"),
+                                },
+                                timeout=1.0,
+                            )
+                        except WorkerError as exc:
+                            if isinstance(exc, WorkerRejected):
+                                if str(exc) == "repository worker is exiting":
+                                    cls._wait_for_exit(existing.pid, runtime_path, process_start=existing.process_start)
+                                    break
+                                if str(exc) != "repository worker controller ownership changed":
+                                    raise
+                            last_error = exc
+                            if not cls._runtime_alive(existing):
+                                break
+                            if attempt < 3:
+                                time.sleep(0.2)
+                            continue
+                        client._status = response["status"]
+                        client._controller_generation = client._status["worker"].get("controller_generation")
+                    needs_restart = existing.priority != priority or existing.version != package_version()
+                    if needs_restart and not client.busy:
+                        client.shutdown()
+                        cls._wait_for_exit(existing.pid, runtime_path, process_start=existing.process_start)
+                        break
+                    return client
+                else:
                     raise WorkerError(
-                        f"repository worker protocol mismatch: {existing.protocol} != {PROTOCOL_VERSION}; "
-                        "restart the server after the active worker has exited"
-                    )
-                runtime_path.unlink(missing_ok=True)
-                existing = None
-
-        if existing is not None:
-            client = cls(runtime_path, existing, desired_priority=priority)
-            try:
-                response = client._rpc({"command": "ping"})
-            except Exception as exc:
-                if cls._pid_alive(existing.pid):
+                        f"repository worker pid {existing.pid} remains unresponsive after four RPC attempts; "
+                        f"identity={existing.process_start or 'legacy/unknown'}, last_error={last_error}; "
+                        "refusing concurrent repository access"
+                    ) from last_error
+                if cls._runtime_alive(existing):
                     raise WorkerError(
-                        f"repository worker pid {existing.pid} is alive but not responding; "
-                        "refusing to start a second worker"
-                    ) from exc
-                runtime_path.unlink(missing_ok=True)
-            else:
-                client._status = response["status"]
-                if Path(existing.storage_root).resolve() != storage_root.resolve():
-                    raise WorkerError("repository worker runtime points at a different storage root")
-
-                needs_restart = existing.priority != priority or existing.version != package_version()
-                can_restart = needs_restart and not client.busy
-                if can_restart:
-                    client.shutdown()
-                    cls._wait_for_exit(existing.pid, runtime_path)
-                else:
-                    return client
+                        f"repository worker pid {existing.pid} is still active after reconnect attempts; "
+                        f"last_error={last_error}; refusing a second writer"
+                    ) from last_error
 
         return cls._start(storage_root, runtime_path, priority)
 
+    @classmethod
+    def _runtime_alive(cls, runtime: RuntimeInfo) -> bool:
+        observed = process_identity(runtime.pid)
+        if runtime.process_start and observed is not None:
+            return observed == runtime.process_start
+        if runtime.process_start and sys.platform.startswith("linux"):
+            return False
+        return cls._pid_alive(runtime.pid)
+
     @classmethod
     def _start(cls, storage_root: Path, runtime_path: Path, priority: str) -> RepositoryWorkerClient:
-        runtime_path.unlink(missing_ok=True)
+        identity = process_identity(os.getpid())
+        launch_id = uuid.uuid4().hex
+        if identity is None:
+            raise WorkerError("cannot identify the current controller process")
         env = os.environ.copy()
         env["PYTHONPATH"] = os.pathsep.join(sys.path)
         env.pop("LD_PRELOAD", None)
@@ -102,6 +156,12 @@ def _start(cls, storage_root: Path, runtime_path: Path, priority: str) -> Reposi
             str(runtime_path),
             "--priority",
             priority,
+            "--controller-pid",
+            str(os.getpid()),
+            "--controller-start",
+            identity,
+            "--launch-id",
+            launch_id,
         ]
         kwargs: dict[str, Any] = {
             "env": env,
@@ -120,19 +180,22 @@ def _start(cls, storage_root: Path, runtime_path: Path, priority: str) -> Reposi
             kwargs["stderr"] = subprocess.STDOUT
             process = subprocess.Popen(command, **kwargs)
 
-        threading.Thread(
-            target=process.wait,
-            name=f"endkeep-worker-reaper-{process.pid}",
-            daemon=True,
-        ).start()
+        if os.name != "nt":
+            threading.Thread(
+                target=process.wait,
+                name=f"endkeep-worker-reaper-{process.pid}",
+                daemon=True,
+            ).start()
 
         deadline = time.monotonic() + cls.START_TIMEOUT_SECONDS
         last_error: Exception | None = None
         while time.monotonic() < deadline:
-            if process.poll() is not None:
-                raise WorkerError(f"repository worker exited during startup with code {process.returncode}")
             runtime = cls._load_runtime(runtime_path)
-            if runtime is not None:
+            if process.poll() is not None and (runtime is None or runtime.launch_id != launch_id):
+                raise WorkerError(f"repository worker launcher exited during startup with code {process.returncode}")
+            if runtime is not None and runtime.launch_id == launch_id:
+                if runtime.process_start and process_identity(runtime.pid) != runtime.process_start:
+                    raise WorkerError("newly launched worker process identity cannot be verified")
                 client = cls(runtime_path, runtime, desired_priority=priority)
                 try:
                     response = client._rpc({"command": "ping"})
@@ -140,11 +203,22 @@ def _start(cls, storage_root: Path, runtime_path: Path, priority: str) -> Reposi
                     last_error = exc
                 else:
                     client._status = response["status"]
+                    client._controller_generation = client._status["worker"].get("controller_generation")
                     return client
             time.sleep(0.05)
 
         process.terminate()
-        raise WorkerError("repository worker did not become ready before the startup deadline") from last_error
+        try:
+            detail = log_path.read_text(encoding="utf-8", errors="replace")[-2500:]
+        except OSError:
+            detail = "<worker log unavailable>"
+        observed = cls._load_runtime(runtime_path)
+        raise WorkerError(
+            f"repository worker did not become ready before the startup deadline; "
+            f"child_pid={process.pid} runtime_exists={runtime_path.exists()} "
+            f"runtime_pid={observed.pid if observed else None} "
+            f"last_rpc_error={last_error!r}; worker log tail: {detail}"
+        ) from last_error
 
     @staticmethod
     def _load_runtime(path: Path) -> RuntimeInfo | None:
@@ -193,11 +267,15 @@ def _pid_alive(pid: int) -> bool:
         return True
 
     @classmethod
-    def _wait_for_exit(cls, pid: int, runtime_path: Path) -> None:
+    def _wait_for_exit(cls, pid: int, runtime_path: Path, *, process_start: str = "") -> None:
         deadline = time.monotonic() + cls.START_TIMEOUT_SECONDS
         while time.monotonic() < deadline:
-            if not cls._pid_alive(pid):
-                runtime_path.unlink(missing_ok=True)
+            observed = process_identity(pid)
+            if (
+                (process_start and observed is not None and observed != process_start)
+                or (process_start and observed is None and sys.platform.startswith("linux"))
+                or not cls._pid_alive(pid)
+            ):
                 return
             time.sleep(0.05)
         raise WorkerError(f"repository worker pid {pid} did not exit cleanly")
@@ -374,6 +452,10 @@ def shutdown(self) -> None:
         self._rpc({"command": "shutdown"})
 
     def detach(self) -> None:
-        # The worker intentionally survives plugin reloads. The next plugin
-        # instance adopts the same runtime through worker-runtime.json.
-        return None
+        # A reload can reclaim the worker within the handoff window; the worker
+        # also watches the BDS process identity to detect an actual server exit.
+        try:
+            self._rpc({"command": "detach", "controller_generation": self._controller_generation}, timeout=0.5)
+        except WorkerRejected as exc:
+            if "unknown worker command" not in str(exc):
+                raise
diff --git a/src/endstone_endkeep/worker/process.py b/src/endstone_endkeep/worker/process.py
new file mode 100644
index 00000000..0b0e9a7f
--- /dev/null
+++ b/src/endstone_endkeep/worker/process.py
@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import os
+import sys
+
+
+def process_identity(pid: int) -> str | None:
+    """Return the OS process creation identity, not just its reusable PID."""
+    if pid <= 0:
+        return None
+    if sys.platform.startswith("linux"):
+        try:
+            stat = open(f"/proc/{pid}/stat", encoding="ascii").read()
+            fields = stat[stat.rfind(")") + 2 :].split()
+            if fields[0] in ("Z", "X"):
+                return None
+            return f"linux:{fields[19]}"
+        except (OSError, ValueError, IndexError):
+            return None
+    if os.name == "nt":
+        import ctypes
+        from ctypes import wintypes
+
+        kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+        kernel32.OpenProcess.argtypes = (wintypes.DWORD, wintypes.BOOL, wintypes.DWORD)
+        kernel32.OpenProcess.restype = wintypes.HANDLE
+        kernel32.GetProcessTimes.argtypes = (
+            wintypes.HANDLE,
+            ctypes.POINTER(wintypes.FILETIME),
+            ctypes.POINTER(wintypes.FILETIME),
+            ctypes.POINTER(wintypes.FILETIME),
+            ctypes.POINTER(wintypes.FILETIME),
+        )
+        kernel32.GetProcessTimes.restype = wintypes.BOOL
+        kernel32.WaitForSingleObject.argtypes = (wintypes.HANDLE, wintypes.DWORD)
+        kernel32.WaitForSingleObject.restype = wintypes.DWORD
+        kernel32.CloseHandle.argtypes = (wintypes.HANDLE,)
+        kernel32.CloseHandle.restype = wintypes.BOOL
+        handle = kernel32.OpenProcess(0x1000 | 0x00100000, False, pid)
+        if not handle:
+            return None
+        try:
+            if kernel32.WaitForSingleObject(handle, 0) != 0x102:
+                return None
+            created = wintypes.FILETIME()
+            exited = wintypes.FILETIME()
+            kernel = wintypes.FILETIME()
+            user = wintypes.FILETIME()
+            if not kernel32.GetProcessTimes(handle, created, exited, kernel, user):
+                return None
+            value = (created.dwHighDateTime << 32) | created.dwLowDateTime
+            return f"windows:{value}"
+        finally:
+            kernel32.CloseHandle(handle)
+    return None
diff --git a/src/endstone_endkeep/worker/protocol.py b/src/endstone_endkeep/worker/protocol.py
index 3708e841..e4434ebc 100644
--- a/src/endstone_endkeep/worker/protocol.py
+++ b/src/endstone_endkeep/worker/protocol.py
@@ -28,6 +28,8 @@ class RuntimeInfo:
     instance_id: str
     storage_root: str
     priority: str
+    process_start: str = ""
+    launch_id: str = ""
 
     def to_dict(self) -> dict:
         return asdict(self)
@@ -43,6 +45,8 @@ def from_dict(cls, raw: dict[str, Any]) -> RuntimeInfo:
             instance_id=str(raw["instance_id"]),
             storage_root=str(raw["storage_root"]),
             priority=str(raw["priority"]),
+            process_start=str(raw.get("process_start", "")),
+            launch_id=str(raw.get("launch_id", "")),
         )
 
     @classmethod
diff --git a/src/endstone_endkeep/worker/server.py b/src/endstone_endkeep/worker/server.py
index 520458a2..33c7a2d7 100644
--- a/src/endstone_endkeep/worker/server.py
+++ b/src/endstone_endkeep/worker/server.py
@@ -10,6 +10,7 @@
 import sys
 import threading
 import time
+import traceback
 import uuid
 from collections import deque
 from dataclasses import asdict
@@ -21,9 +22,11 @@
 from endstone_endkeep.repository.progress import ProgressTracker
 from endstone_endkeep.repository.recovery import StartupRecovery
 
+from .process import process_identity
 from .protocol import PROTOCOL_VERSION, RuntimeInfo, package_version, receive_message, send_message
 
 _IDLE_EXIT_SECONDS = 300.0
+_HANDOFF_GRACE_SECONDS = 15.0
 _PRIORITY_NICE = {
     "conservative": 15,
     "background": 8,
@@ -38,9 +41,16 @@ class _ThreadingServer(socketserver.ThreadingTCPServer):
 
 
 class WorkerApplication:
-    def __init__(self, storage_root: Path, *, priority: str) -> None:
+    def __init__(
+        self, storage_root: Path, *, priority: str, controller_pid: int = 0, controller_start: str = ""
+    ) -> None:
         self.storage_root = storage_root
         self.priority = priority
+        self.controller_pid = controller_pid
+        self.controller_start = controller_start
+        self.controller_generation = 0
+        self.detached_until: float | None = None
+        self._results_path = storage_root / "worker-results.json"
         self.tracker = ProgressTracker()
         self.service: RepositoryService | None = None
         self.settings: dict[str, Any] | None = None
@@ -48,10 +58,13 @@ def __init__(self, storage_root: Path, *, priority: str) -> None:
         self.results: deque[dict[str, Any]] = deque()
         self._next_result_id = 1
         self._last_acked_result_id = 0
+        self._results_dirty = False
+        self._load_results()
         self._active_request_id: str | None = None
         self._management_requests: dict[str, tuple[str, ...]] = {}
         self.last_contact = time.monotonic()
         self.shutdown_requested = False
+        self._stopping = False
         self._lock = threading.RLock()
         self._stop_callback = lambda: None
 
@@ -62,8 +75,40 @@ def handle(self, request: dict[str, Any]) -> dict[str, Any]:
         with self._lock:
             self.last_contact = time.monotonic()
             command = request.get("command")
+            if self._stopping and command not in ("ping", "status", "poll", "ack_result", "shutdown"):
+                return {"ok": False, "error": "repository worker is exiting"}
             if command == "ping":
                 return {"ok": True, "status": self._status_locked()}
+            if command == "attach":
+                if request.get("controller_generation") != self.controller_generation:
+                    return {"ok": False, "error": "repository worker controller ownership changed"}
+                pid = int(request.get("controller_pid", 0))
+                started = str(request.get("controller_start", ""))
+                if not started or process_identity(pid) != started:
+                    return {"ok": False, "error": "controller process identity cannot be verified"}
+                controller_alive = bool(self.controller_start) and (
+                    process_identity(self.controller_pid) == self.controller_start
+                )
+                if self.controller_start and not controller_alive:
+                    self._request_stop_locked()
+                    return {"ok": False, "error": "repository worker is exiting"}
+                if (
+                    self.controller_pid
+                    and (pid, started) != (self.controller_pid, self.controller_start)
+                    and controller_alive
+                ):
+                    return {"ok": False, "error": "repository worker belongs to a live controller"}
+                self.controller_pid = pid
+                self.controller_start = started
+                self.controller_generation += 1
+                self.detached_until = None
+                self.shutdown_requested = False
+                return {"ok": True, "status": self._status_locked()}
+            if command == "detach":
+                if request.get("controller_generation") != self.controller_generation:
+                    return {"ok": False, "error": "repository worker controller ownership changed"}
+                self.detached_until = time.monotonic() + _HANDOFF_GRACE_SECONDS
+                return {"ok": True, "status": self._status_locked()}
             if command == "configure":
                 return self._configure_locked(request.get("settings"))
             if command == "status":
@@ -117,6 +162,7 @@ def handle(self, request: dict[str, Any]) -> dict[str, Any]:
                 elif self.results and int(self.results[0]["id"]) == result_id:
                     self.results.popleft()
                     self._last_acked_result_id = result_id
+                    self._results_dirty = True
                     accepted = True
                 else:
                     accepted = False
@@ -265,7 +311,14 @@ def _apply_settings_locked(self, settings: dict[str, Any]) -> dict[str, Any]:
 
     def _ready_for_new_job_locked(self) -> bool:
         service = self.service
-        return service is not None and not service.busy and not self.results and self.pending_settings is None
+        return (
+            not self.shutdown_requested
+            and not self._stopping
+            and service is not None
+            and not service.busy
+            and not self.results
+            and self.pending_settings is None
+        )
 
     def _require_service(self) -> RepositoryService:
         if self.service is None:
@@ -280,6 +333,9 @@ def _status_locked(self, *, include_repository: bool = False) -> dict[str, Any]:
                 "pid": os.getpid(),
                 "priority": self.priority,
                 "alive": True,
+                "controller_pid": self.controller_pid,
+                "controller_start": self.controller_start,
+                "controller_generation": self.controller_generation,
             },
             "job": self.tracker.snapshot().to_dict(),
             "job_occupied": service is not None and service.busy,
@@ -327,6 +383,27 @@ def _repository_status_locked(self) -> dict[str, Any]:
         }
 
     def monitor(self) -> None:
+        try:
+            self._monitor()
+        except Exception:
+            with self._lock:
+                self.shutdown_requested = True
+                self._stopping = True
+            try:
+                traceback.print_exc()
+            finally:
+                self._stop_callback()
+
+    def _request_stop_locked(self) -> None:
+        self.shutdown_requested = True
+        self._stopping = True
+        if self.service is not None:
+            self.service.request_cancel()
+
+    def _monitor(self) -> None:
+        next_owner_check = 0.0
+        last_owner: tuple[int, str] | None = None
+        controller_gone = False
         while True:
             time.sleep(0.1)
             with self._lock:
@@ -354,27 +431,93 @@ def monitor(self) -> None:
                     service = self.service
 
                 busy = service is not None and service.busy
-                idle_too_long = time.monotonic() - self.last_contact >= _IDLE_EXIT_SECONDS
-                should_stop = self.shutdown_requested or (idle_too_long and not busy and not self.results)
+                now = time.monotonic()
+                owner = (self.controller_pid, self.controller_start)
+                if now >= next_owner_check or owner != last_owner:
+                    controller_gone = (
+                        bool(self.controller_start) and process_identity(self.controller_pid) != self.controller_start
+                    )
+                    last_owner = owner
+                    next_owner_check = now + 1.0
+                dirty_ack = (list(self.results), self._last_acked_result_id) if self._results_dirty else None
+                handoff_expired = self.detached_until is not None and now >= self.detached_until
+                idle_too_long = now - self.last_contact >= _IDLE_EXIT_SECONDS
+                if controller_gone or handoff_expired:
+                    self._request_stop_locked()
+                should_stop = not busy and (
+                    self.shutdown_requested or controller_gone or handoff_expired or idle_too_long
+                )
+                if should_stop:
+                    self._stopping = True
+            if dirty_ack is not None:
+                results, acked = dirty_ack
+                self._save_results(results, acked)
+                with self._lock:
+                    if acked == self._last_acked_result_id and results == list(self.results):
+                        self._results_dirty = False
             if should_stop:
                 self._stop_callback()
                 return
 
     def _queue_result(self, payload: dict[str, Any]) -> None:
-        self.results.append(
-            {
-                "id": self._next_result_id,
-                "payload": payload,
-            }
-        )
+        result = {"id": self._next_result_id, "payload": payload}
+        self.results.append(result)
         self._next_result_id += 1
+        self._results_dirty = True
+        self._save_results(list(self.results), self._last_acked_result_id)
+        self._results_dirty = False
+
+    def _load_results(self) -> None:
+        try:
+            data = json.loads(self._results_path.read_text(encoding="utf-8"))
+        except FileNotFoundError:
+            return
+        if not isinstance(data, dict) or not isinstance(data.get("results"), list):
+            raise ValueError("invalid persisted repository worker results")
+        results = data["results"]
+        if any(not isinstance(item, dict) or "id" not in item or "payload" not in item for item in results):
+            raise ValueError("invalid persisted repository worker result envelope")
+        self.results = deque(results)
+        self._last_acked_result_id = int(data.get("last_acked", 0))
+        self._next_result_id = max([self._last_acked_result_id, *(int(item["id"]) for item in results)]) + 1
+
+    def _save_results(self, results: list[dict[str, Any]], last_acked: int) -> None:
+        self._results_path.parent.mkdir(parents=True, exist_ok=True)
+        tmp = self._results_path.with_name(f".{self._results_path.name}.{os.getpid()}.tmp")
+        try:
+            with tmp.open("w", encoding="utf-8") as stream:
+                json.dump({"results": results, "last_acked": last_acked}, stream)
+                stream.flush()
+                os.fsync(stream.fileno())
+            os.replace(tmp, self._results_path)
+        finally:
+            tmp.unlink(missing_ok=True)
 
     def close(self) -> None:
         with self._lock:
             service = self.service
-            self.service = None
-        if service is not None:
-            service.close()
+            self._stopping = True
+        try:
+            if service is not None:
+                service.close()
+                try:
+                    result = service.poll()
+                except Exception as exc:
+                    payload = {"kind": "error", "error": str(exc)}
+                else:
+                    payload = asdict(result) if result is not None else None
+                if payload is not None:
+                    with self._lock:
+                        if self._active_request_id is not None:
+                            payload["request_id"] = self._active_request_id
+                        self._queue_result(payload)
+                        self._active_request_id = None
+        finally:
+            with self._lock:
+                self.service = None
+                if self._results_dirty:
+                    self._save_results(list(self.results), self._last_acked_result_id)
+                    self._results_dirty = False
 
 
 class _RequestHandler(socketserver.BaseRequestHandler):
@@ -472,6 +615,9 @@ def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description="EndKeep repository worker")
     parser.add_argument("--storage-root", required=True)
     parser.add_argument("--runtime-file", required=True)
+    parser.add_argument("--controller-pid", type=int, default=0)
+    parser.add_argument("--controller-start", default="")
+    parser.add_argument("--launch-id", required=True)
     parser.add_argument(
         "--priority",
         required=True,
@@ -487,9 +633,20 @@ def main(argv: list[str] | None = None) -> int:
     storage_root.mkdir(parents=True, exist_ok=True)
     _apply_priority(args.priority)
 
+    with RepositoryLock(storage_root / "worker-owner"):
+        return _run_worker(
+            storage_root, runtime_file, args.priority, args.controller_pid, args.controller_start, args.launch_id
+        )
+
+
+def _run_worker(
+    storage_root: Path, runtime_file: Path, priority: str, controller_pid: int, controller_start: str, launch_id: str
+) -> int:
     token = secrets.token_hex(32)
     instance_id = uuid.uuid4().hex
-    app = WorkerApplication(storage_root, priority=args.priority)
+    app = WorkerApplication(
+        storage_root, priority=priority, controller_pid=controller_pid, controller_start=controller_start
+    )
 
     with _ThreadingServer(("127.0.0.1", 0), _RequestHandler) as server:
         server.app = app  # type: ignore[attr-defined]
@@ -503,7 +660,9 @@ def main(argv: list[str] | None = None) -> int:
             token=token,
             instance_id=instance_id,
             storage_root=str(storage_root),
-            priority=args.priority,
+            priority=priority,
+            process_start=process_identity(os.getpid()) or "",
+            launch_id=launch_id,
         )
         _write_runtime(runtime_file, runtime)
 
@@ -512,6 +671,8 @@ def main(argv: list[str] | None = None) -> int:
         try:
             server.serve_forever(poll_interval=0.2)
         finally:
-            app.close()
-            _remove_runtime(runtime_file, instance_id)
+            try:
+                app.close()
+            finally:
+                _remove_runtime(runtime_file, instance_id)
     return 0

From 5051a80cb9ad7c26e4338eec3bab0786bb327d60 Mon Sep 17 00:00:00 2001
From: ReallocAll <ReallocAll@outlook.com>
Date: Thu, 15 Oct 2026 10:27:21 +0800
Subject: [PATCH 2/4] fix: cancel repository work safely and preserve
 recoverable transactions

---
 src/endstone_endkeep/logical/diff.py          |  7 ++++
 src/endstone_endkeep/logical/merge.py         |  7 ++++
 src/endstone_endkeep/logical/sidecar.py       | 13 +++++-
 src/endstone_endkeep/offline/cli.py           | 36 ++++++++++++----
 src/endstone_endkeep/repository/logicalize.py | 23 +++++++---
 .../repository/maintenance.py                 | 42 ++++++++++++-------
 src/endstone_endkeep/repository/mutation.py   | 25 ++++++++---
 src/endstone_endkeep/repository/objects.py    | 24 +++++++----
 src/endstone_endkeep/repository/reader.py     |  5 ++-
 src/endstone_endkeep/repository/rollover.py   |  5 ++-
 src/endstone_endkeep/repository/verify.py     | 11 +++--
 src/endstone_endkeep/staging/clone.py         | 17 +++++++-
 12 files changed, 164 insertions(+), 51 deletions(-)

diff --git a/src/endstone_endkeep/logical/diff.py b/src/endstone_endkeep/logical/diff.py
index f73d7cf4..63334474 100644
--- a/src/endstone_endkeep/logical/diff.py
+++ b/src/endstone_endkeep/logical/diff.py
@@ -121,6 +121,8 @@ def compose_delta_pair(
     first: Iterable[DeltaOperation],
     second: Iterable[DeltaOperation],
     stats: DiffStats,
+    *,
+    progress: Callable[[int], None] | None = None,
 ) -> Iterator[DeltaOperation]:
     """Build the minimal previous-to-successor DELTA in one bounded-memory pass.
 
@@ -136,8 +138,13 @@ def compose_delta_pair(
     first_item = next(first_it, missing)
     second_item = next(second_it, missing)
     last_key: bytes | None = None
+    processed = 0
 
     while previous_item is not missing or first_item is not missing or second_item is not missing:
+        if progress is not None:
+            processed += 1
+            if processed % 16384 == 0:
+                progress(16384)
         keys = []
         if previous_item is not missing:
             keys.append(previous_item[0])
diff --git a/src/endstone_endkeep/logical/merge.py b/src/endstone_endkeep/logical/merge.py
index b977138a..4b0c42a1 100644
--- a/src/endstone_endkeep/logical/merge.py
+++ b/src/endstone_endkeep/logical/merge.py
@@ -17,13 +17,20 @@ class MergeError(ValueError):
 def apply_delta(
     state: Iterable[tuple[bytes, bytes]],
     delta: Iterable[DeltaOperation],
+    *,
+    cancel_check: Callable[[], None] | None = None,
 ) -> Iterator[tuple[bytes, bytes]]:
     state_it = iter(state)
     delta_it = iter(delta)
     state_item = next(state_it, _SENTINEL)
     operation = next(delta_it, _SENTINEL)
+    consumed = 0
 
     while state_item is not _SENTINEL or operation is not _SENTINEL:
+        if cancel_check is not None:
+            consumed += 1
+            if consumed % 16384 == 0:
+                cancel_check()
         if operation is _SENTINEL:
             yield state_item
             state_item = next(state_it, _SENTINEL)
diff --git a/src/endstone_endkeep/logical/sidecar.py b/src/endstone_endkeep/logical/sidecar.py
index ef0958de..ffad503d 100644
--- a/src/endstone_endkeep/logical/sidecar.py
+++ b/src/endstone_endkeep/logical/sidecar.py
@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import os
-from collections.abc import Iterable
+from collections.abc import Callable, Iterable
 from dataclasses import dataclass
 from pathlib import Path, PurePosixPath
 from typing import BinaryIO
@@ -23,6 +23,8 @@ def write_sidecar(
     stream: BinaryIO,
     raw_snapshot_root: Path,
     entries: Iterable[SnapshotEntry],
+    *,
+    cancel_check: Callable[[], None] | None = None,
 ) -> SidecarStats:
     stream.write(SIDECAR_MAGIC)
     ordered = sorted(entries, key=lambda entry: entry.path.as_posix())
@@ -31,6 +33,8 @@ def write_sidecar(
     total_bytes = 0
 
     for entry in ordered:
+        if cancel_check is not None:
+            cancel_check()
         path_text = entry.path.as_posix()
         if previous is not None and path_text <= previous:
             raise LogicalFormatError("SIDECAR paths must be strictly increasing")
@@ -43,6 +47,8 @@ def write_sidecar(
         with path.open("rb") as source:
             remaining = entry.snapshot_bytes
             while remaining:
+                if cancel_check is not None:
+                    cancel_check()
                 chunk = source.read(min(1024 * 1024, remaining))
                 if not chunk:
                     raise LogicalFormatError(f"SIDECAR source is short: {entry.path}")
@@ -65,6 +71,7 @@ def extract_sidecar(
     destination_root: Path,
     *,
     strip_prefix: str | None = None,
+    cancel_check: Callable[[], None] | None = None,
 ) -> SidecarStats:
     if _read_exact(stream, len(SIDECAR_MAGIC)) != SIDECAR_MAGIC:
         raise LogicalFormatError("invalid SIDECAR magic/version")
@@ -75,6 +82,8 @@ def extract_sidecar(
     total_bytes = 0
 
     while True:
+        if cancel_check is not None:
+            cancel_check()
         first = stream.read(1)
         if not first:
             break
@@ -117,6 +126,8 @@ def extract_sidecar(
         try:
             remaining = size
             while remaining:
+                if cancel_check is not None:
+                    cancel_check()
                 chunk = stream.read(min(1024 * 1024, remaining))
                 if not chunk:
                     raise LogicalFormatError(f"truncated SIDECAR file: {path_text!r}")
diff --git a/src/endstone_endkeep/offline/cli.py b/src/endstone_endkeep/offline/cli.py
index 2529106b..418c119b 100644
--- a/src/endstone_endkeep/offline/cli.py
+++ b/src/endstone_endkeep/offline/cli.py
@@ -129,6 +129,7 @@ def _verify_restore_dependencies(
     *,
     verbose: bool,
     show_progress: bool,
+    cancel_check: Callable[[], None] | None = None,
 ) -> None:
     """Deep-verify only the immutable objects required to restore one snapshot."""
 
@@ -151,7 +152,7 @@ def _verify_restore_dependencies(
                 f"logical_bytes={metadata.logical_bytes} compressed_bytes={metadata.compressed_bytes}",
                 enabled=verbose,
             )
-            objects.verify(metadata)
+            objects.verify(metadata, cancel_check=cancel_check)
             if bar is not None:
                 bar.update(1)
             _verbose(
@@ -391,7 +392,10 @@ def command_restore(
     preflight: Callable[[SnapshotNode], None] | None = None,
     space_guard: Callable[[], None] | None = None,
     progress: Callable[[int, int], None] | None = None,
+    cancel_check: Callable[[], None] | None = None,
 ) -> int:
+    if cancel_check is not None:
+        cancel_check()
     if destination.exists():
         raise FileExistsError(f"restore destination already exists: {destination}")
 
@@ -417,6 +421,7 @@ def command_restore(
             node,
             verbose=verbose,
             show_progress=show_progress,
+            cancel_check=cancel_check,
         )
 
         destination.parent.mkdir(parents=True, exist_ok=True)
@@ -427,7 +432,7 @@ def command_restore(
                 print(file=sys.stderr)
                 _stage(2, 5, "Restoring SIDECAR")
             with objects.open_logical(node.sidecar) as stream:
-                sidecar = extract_sidecar(stream, destination, strip_prefix=node.world_name)
+                sidecar = extract_sidecar(stream, destination, strip_prefix=node.world_name, cancel_check=cancel_check)
             _verbose(
                 f"SIDECAR files={sidecar.files} bytes={sidecar.bytes} elapsed={time.perf_counter() - started:.3f}s",
                 enabled=verbose,
@@ -437,7 +442,7 @@ def command_restore(
 
             if space_guard is not None:
                 space_guard()
-            state = reader.iter_state(manifest, snapshot=node.snapshot)
+            state = reader.iter_state(manifest, snapshot=node.snapshot, cancel_check=cancel_check)
             if show_progress:
                 print(file=sys.stderr)
                 _stage(3, 5, "Rebuilding LevelDB")
@@ -456,6 +461,8 @@ def command_restore(
 
             def on_written(amount: int) -> None:
                 nonlocal completed_records
+                if cancel_check is not None:
+                    cancel_check()
                 completed_records += amount
                 if rebuild_bar is not None:
                     rebuild_bar.update(amount)
@@ -464,7 +471,9 @@ def on_written(amount: int) -> None:
                 if space_guard is not None:
                     space_guard()
 
-            callback_enabled = rebuild_bar is not None or progress is not None or space_guard is not None
+            callback_enabled = (
+                rebuild_bar is not None or progress is not None or space_guard is not None or cancel_check is not None
+            )
             try:
                 records, value_bytes = write_fresh_leveldb(
                     destination / "db",
@@ -498,11 +507,18 @@ def on_written(amount: int) -> None:
                 if show_progress
                 else None
             )
+
+            def on_verified(amount: int) -> None:
+                if cancel_check is not None:
+                    cancel_check()
+                if verify_bar is not None:
+                    verify_bar.update(amount)
+
             try:
                 with iter_visible_state(destination / "db") as reopened:
                     reopened_stats = hash_state(
                         reopened,
-                        progress=verify_bar.update if verify_bar is not None else None,
+                        progress=on_verified if verify_bar is not None or cancel_check is not None else None,
                     )
             finally:
                 if verify_bar is not None:
@@ -525,7 +541,9 @@ def on_written(amount: int) -> None:
                 print(file=sys.stderr)
                 _stage(5, 5, "Flushing restored world")
             started = time.perf_counter()
-            _fsync_tree(destination)
+            if cancel_check is not None:
+                cancel_check()
+            _fsync_tree(destination, cancel_check=cancel_check)
             _verbose(
                 f"fsync destination={destination} elapsed={time.perf_counter() - started:.3f}s",
                 enabled=verbose,
@@ -544,12 +562,14 @@ def on_written(amount: int) -> None:
     return 0
 
 
-def _fsync_tree(root: Path) -> None:
+def _fsync_tree(root: Path, *, cancel_check: Callable[[], None] | None = None) -> None:
     directories: list[Path] = []
     for current, _dirnames, filenames in os.walk(root):
         current_path = Path(current)
         directories.append(current_path)
         for filename in filenames:
+            if cancel_check is not None:
+                cancel_check()
             # Windows' CRT _commit needs a writable handle to flush file data.
             # Do not skip fsync: failed flushes must abort recovery.
             flags = os.O_RDWR if os.name == "nt" else os.O_RDONLY
@@ -559,6 +579,8 @@ def _fsync_tree(root: Path) -> None:
             finally:
                 os.close(fd)
     for directory in reversed(directories):
+        if cancel_check is not None:
+            cancel_check()
         _fsync_directory(directory)
     _fsync_directory(root.parent)
 
diff --git a/src/endstone_endkeep/repository/logicalize.py b/src/endstone_endkeep/repository/logicalize.py
index 5e83a77e..6629b20e 100644
--- a/src/endstone_endkeep/repository/logicalize.py
+++ b/src/endstone_endkeep/repository/logicalize.py
@@ -15,6 +15,7 @@
 
 from .manifest import ManifestStore, RepositoryManifest, SnapshotNode, snapshot_order_key
 from .objects import ObjectMetadata, ObjectStore
+from .progress import JobCancelled
 from .reader import RepositoryReader
 from .transaction import RepositoryTransaction
 
@@ -89,10 +90,11 @@ def logicalize(
                 raise ValueError(f"raw snapshot {raw.snapshot_id} is not newer than repository tail {tail.snapshot}")
 
         self._report(progress, "clone", None, None, None, raw.snapshot_id)
-        clone_world(raw.path, raw.manifest.world_name, self.work_root, raw.snapshot_id)
+        clone_world(raw.path, raw.manifest.world_name, self.work_root, raw.snapshot_id, cancel_check=cancel_check)
         self._checkpoint(cancel_check)
 
         db_path = self.work_root / raw.snapshot_id / raw.manifest.world_name / "db"
+        cancelled = False
         try:
             if current_manifest is None:
                 node, diff_counts = self._create_base(
@@ -124,8 +126,14 @@ def logicalize(
             new_manifest = RepositoryManifest(generation=generation, chain=tuple(new_chain))
             self._report(progress, "commit", None, None, None, raw.snapshot_id)
             RepositoryTransaction(self.manifests).commit(new_manifest)
+            self._report(progress, "committed", None, None, None, raw.snapshot_id)
+            self._checkpoint(cancel_check)
+        except JobCancelled:
+            cancelled = True
+            raise
         finally:
-            remove_processing_clone(self.work_root, raw.snapshot_id)
+            if not cancelled:
+                remove_processing_clone(self.work_root, raw.snapshot_id)
 
         raw_deleted = True
         try:
@@ -171,7 +179,8 @@ def advance(amount: int) -> None:
                     stream,
                     current,
                     progress=advance if progress is not None or cancel_check is not None else None,
-                )
+                ),
+                cancel_check=cancel_check,
             )
         assert isinstance(stats, BaseStats)
         self._checkpoint(cancel_check)
@@ -205,7 +214,7 @@ def _create_delta(
         cancel_check: CancelCheck | None,
     ) -> tuple[SnapshotNode, tuple[int, int, int, int]]:
         stats = DiffStats()
-        previous = self.reader.iter_state(current_manifest)
+        previous = self.reader.iter_state(current_manifest, cancel_check=cancel_check)
         previous_total = current_manifest.chain[-1].records
 
         def advance(_amount: int) -> None:
@@ -238,7 +247,8 @@ def advance(_amount: int) -> None:
                         stats,
                         progress=advance if progress is not None or cancel_check is not None else None,
                     ),
-                )
+                ),
+                cancel_check=cancel_check,
             )
         assert isinstance(delta_stats, DeltaStats)
         if delta_stats.records != stats.delta_records:
@@ -275,7 +285,8 @@ def _create_sidecar(
         self._report(progress, "sidecar", None, None, None, raw.snapshot_id)
         self._checkpoint(cancel_check)
         metadata, stats = self.objects.create(
-            lambda stream: write_sidecar(stream, raw.path, raw.manifest.sidecar_entries)
+            lambda stream: write_sidecar(stream, raw.path, raw.manifest.sidecar_entries, cancel_check=cancel_check),
+            cancel_check=cancel_check,
         )
         self._checkpoint(cancel_check)
         return metadata, stats
diff --git a/src/endstone_endkeep/repository/maintenance.py b/src/endstone_endkeep/repository/maintenance.py
index 3dea95fc..931bd5ec 100644
--- a/src/endstone_endkeep/repository/maintenance.py
+++ b/src/endstone_endkeep/repository/maintenance.py
@@ -17,6 +17,7 @@
 from .progress import JobCancelled, ProgressTracker
 from .raw_queue import RawLimitResult, RawQueue
 from .reader import RepositoryReader
+from .recovery import StartupRecovery
 from .retention import RetentionDecision, select_retention
 from .rollover import Rollover, RolloverResult
 from .verify import RepositoryVerifier, VerifyReport
@@ -95,6 +96,7 @@ def __init__(
         self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="endkeep-repository")
         self._future: Future[RepositoryJobResult] | None = None
         self._closed = False
+        self._recover_after_cancel = False
 
     @property
     def busy(self) -> bool:
@@ -181,14 +183,13 @@ def start_mutation(self, operation: str, snapshot: str, expected_generation: int
             expected_generation,
             mode=operation,
         )
-        # SnapshotMutator owns its atomic commit sequence; it has no cooperative checkpoints.
-        self.tracker.update(cancelable=False)
         return accepted
 
     def _run_mutation(self, operation: str, snapshot: str, expected_generation: int) -> RepositoryJobResult:
         with RepositoryLock(self.repo_root):
             self.health.require_healthy()
-            self.tracker.update(stage="Validate", detail="checking repository generation", cancelable=False)
+            self.tracker.update(stage="Validate", detail="checking repository generation", cancelable=True)
+            self._checkpoint()
             manifest = self.manifests.load_current()
             if manifest is None:
                 raise ValueError("repository is empty")
@@ -198,12 +199,18 @@ def _run_mutation(self, operation: str, snapshot: str, expected_generation: int)
                 )
             # A previous interrupted commit may have left unpublished generation files.
             self.manifests.discard_unpublished()
-            self.tracker.update(stage="Rewrite", detail=f"{operation} {snapshot}", cancelable=False)
-            mutator = SnapshotMutator(self.manifests, self.objects)
+            self.tracker.update(stage="Rewrite", detail=f"{operation} {snapshot}", cancelable=True)
+
+            def phase(name: str) -> None:
+                self._checkpoint()
+                self.tracker.update(cancelable=name != "commit")
+
+            mutator = SnapshotMutator(self.manifests, self.objects, cancel_check=self._checkpoint, phase=phase)
             updated = (
                 mutator.delete(manifest, snapshot) if operation == "delete" else mutator.rollover(manifest, snapshot)
             )
-            self.tracker.update(stage="Finalize", detail="manifest committed", cancelable=False)
+            self.tracker.update(stage="Finalize", detail="manifest committed", cancelable=True)
+            self._checkpoint()
             result = {
                 "operation": operation,
                 "snapshot": snapshot,
@@ -220,7 +227,6 @@ def start_export(self, snapshot: str) -> bool:
         if not re.fullmatch(r"[0-9]{8}-[0-9]{6}(?:-[0-9]{2})?", snapshot):
             raise ValueError("export requires an exact snapshot ID (YYYYMMDD-HHMMSS)")
         accepted = self._submit("export", ("Restore", "Finalize"), self._run_export, snapshot)
-        self.tracker.update(cancelable=False)
         return accepted
 
     def _run_export(self, snapshot: str) -> RepositoryJobResult:
@@ -268,14 +274,15 @@ def progress(current: int, total: int) -> None:
                 total=total,
                 unit="records",
                 detail="rebuilding a separate LevelDB",
-                cancelable=False,
+                cancelable=True,
             )
+            self._checkpoint()
 
         self.tracker.update(
             stage="Restore",
             snapshot=snapshot,
             detail="verifying and rebuilding a separate world",
-            cancelable=False,
+            cancelable=True,
         )
         # The shared restore owns the repository lock and verifies the restored
         # state digest; failures remove the newly created destination.
@@ -287,17 +294,14 @@ def progress(current: int, total: int) -> None:
             preflight=preflight,
             space_guard=space_guard,
             progress=progress,
+            cancel_check=self._checkpoint,
         )
-        self.tracker.update(stage="Finalize", detail="export complete", cancelable=False)
+        self.tracker.update(stage="Finalize", detail="export complete", cancelable=True)
         return RepositoryJobResult(kind="export", export={"snapshot": snapshot, "destination": str(destination)})
 
     def request_cancel(self) -> bool:
         if self._closed or not self.busy:
             return False
-        # These operations have no cooperative checkpoints: never acknowledge a
-        # cancellation that cannot be honored.
-        if self.tracker.snapshot().kind in ("mutation", "export"):
-            return False
         return self.tracker.request_cancel()
 
     def poll(self) -> RepositoryJobResult | None:
@@ -325,8 +329,15 @@ def _submit(
 
     def _execute_job(self, kind: str, function, *args) -> RepositoryJobResult:
         try:
+            if self._recover_after_cancel:
+                with RepositoryLock(self.repo_root):
+                    StartupRecovery(self.storage_root, self.manifests, self.objects).run(
+                        preserve_committed_raw=self.health.failed
+                    )
+                self._recover_after_cancel = False
             return function(*args)
         except JobCancelled:
+            self._recover_after_cancel = True
             return RepositoryJobResult(kind="cancelled", cancelled_kind=kind)
         finally:
             self.tracker.finish()
@@ -349,6 +360,7 @@ def _logical_progress(
             "diff+compress": "Logicalize",
             "sidecar": "Sidecar",
             "commit": "Commit",
+            "committed": "Finalize",
         }.get(detail, "Logicalize")
         committing = stage == "Commit"
         self.tracker.update(
@@ -397,7 +409,7 @@ def logicalize(path: Path) -> None:
 
     def _checked_verify(self, verifier: RepositoryVerifier, *, deep: bool, **kwargs) -> VerifyReport:
         try:
-            report = verifier.verify(deep=deep, **kwargs)
+            report = verifier.verify(deep=deep, cancel_check=self._checkpoint, **kwargs)
         except JobCancelled:
             raise
         except Exception as exc:
diff --git a/src/endstone_endkeep/repository/mutation.py b/src/endstone_endkeep/repository/mutation.py
index 87d9aaf2..69cf6c4c 100644
--- a/src/endstone_endkeep/repository/mutation.py
+++ b/src/endstone_endkeep/repository/mutation.py
@@ -24,11 +24,15 @@ def __init__(
         objects: ObjectStore,
         *,
         fault_hook: Callable[[str], None] | None = None,
+        cancel_check: Callable[[], None] | None = None,
+        phase: Callable[[str], None] | None = None,
     ) -> None:
         self.manifests = manifests
         self.objects = objects
         self.reader = RepositoryReader(objects)
         self.fault_hook = fault_hook
+        self.cancel_check = cancel_check
+        self.phase = phase
 
     def delete(self, manifest: RepositoryManifest, snapshot: str) -> RepositoryManifest:
         self._assert_current(manifest)
@@ -50,7 +54,9 @@ def delete(self, manifest: RepositoryManifest, snapshot: str) -> RepositoryManif
             # key-wise merge avoids replaying the old chain twice and never
             # buffers the full world state, including when most keys disappear.
             with ExitStack() as stack:
-                predecessor = self.reader.iter_state(manifest, snapshot=previous.snapshot)
+                predecessor = self.reader.iter_state(
+                    manifest, snapshot=previous.snapshot, cancel_check=self.cancel_check
+                )
                 stack.callback(predecessor.close)
                 removed_stream = stack.enter_context(self.objects.open_logical(manifest.chain[index].object))
                 successor_stream = stack.enter_context(self.objects.open_logical(successor.object))
@@ -62,8 +68,10 @@ def delete(self, manifest: RepositoryManifest, snapshot: str) -> RepositoryManif
                             iter_delta(removed_stream),
                             iter_delta(successor_stream),
                             stats,
+                            progress=(lambda _amount: self.cancel_check()) if self.cancel_check is not None else None,
                         ),
-                    )
+                    ),
+                    cancel_check=self.cancel_check,
                 )
             if delta_stats.records != stats.delta_records:
                 raise RuntimeError("bridge DELTA operation count does not match semantic diff stats")
@@ -73,7 +81,7 @@ def delete(self, manifest: RepositoryManifest, snapshot: str) -> RepositoryManif
                 stats.current_records,
                 stats.current_value_bytes,
             )
-            self.objects.verify(object_meta)
+            self.objects.verify(object_meta, cancel_check=self.cancel_check)
 
             bridge = replace(successor, object=object_meta)
             updated = RepositoryManifest(
@@ -82,9 +90,16 @@ def delete(self, manifest: RepositoryManifest, snapshot: str) -> RepositoryManif
             )
 
             # Replay the newly built chain, not the original successor delta.
-            replay = hash_state(self.reader.iter_state(updated, snapshot=successor.snapshot))
+            replay = hash_state(
+                self.reader.iter_state(updated, snapshot=successor.snapshot, cancel_check=self.cancel_check),
+                progress=(lambda _amount: self.cancel_check()) if self.cancel_check is not None else None,
+            )
             self._check_stats(successor, replay.sha256, replay.records, replay.value_bytes)
 
+        if self.cancel_check is not None:
+            self.cancel_check()
+        if self.phase is not None:
+            self.phase("commit")
         RepositoryTransaction(self.manifests, fault_hook=self.fault_hook).commit(updated)
         return updated
 
@@ -100,7 +115,7 @@ def _rollover(self, manifest: RepositoryManifest, index: int) -> RepositoryManif
             self.manifests,
             self.objects,
             fault_hook=self.fault_hook,
-        ).run(manifest, index)
+        ).run(manifest, index, cancel_check=self.cancel_check, phase=self.phase)
         # Rollover verifies the streamed BASE identity before publication.
         return updated
 
diff --git a/src/endstone_endkeep/repository/objects.py b/src/endstone_endkeep/repository/objects.py
index 320b8e6a..aeac13d6 100644
--- a/src/endstone_endkeep/repository/objects.py
+++ b/src/endstone_endkeep/repository/objects.py
@@ -156,7 +156,9 @@ def prepare(self) -> None:
         if objects_created or incoming_created:
             self._fsync_directory(self.repo_root)
 
-    def create(self, builder: Callable[[BinaryIO], _T]) -> tuple[ObjectMetadata, _T]:
+    def create(
+        self, builder: Callable[[BinaryIO], _T], *, cancel_check: Callable[[], None] | None = None
+    ) -> tuple[ObjectMetadata, _T]:
         """Stream one canonical logical object, verify it, then publish it immutably."""
 
         self.prepare()
@@ -170,13 +172,13 @@ def create(self, builder: Callable[[BinaryIO], _T]) -> tuple[ObjectMetadata, _T]
             result = builder(writer)
             metadata = writer.finish()
             self.fault_hook("after_object_fsync")
-            self.verify_path(incoming, metadata)
+            self.verify_path(incoming, metadata, cancel_check=cancel_check)
 
             target = self.path_for(metadata.logical_sha256)
             shard_created = not target.parent.exists()
             target.parent.mkdir(parents=True, exist_ok=True)
             if target.exists():
-                existing = self._inspect_path(target)
+                existing = self._inspect_path(target, cancel_check=cancel_check)
                 if existing[0] != metadata.logical_sha256 or existing[2] != metadata.logical_bytes:
                     raise ObjectStoreError(
                         "logical object path contains different logical content; refusing to overwrite immutable object"
@@ -223,12 +225,14 @@ def open_logical(self, metadata: ObjectMetadata) -> Generator[BinaryIO]:
             reader.close()
             source.close()
 
-    def verify(self, metadata: ObjectMetadata) -> None:
-        self.verify_path(self.path_for(metadata.logical_sha256), metadata)
+    def verify(self, metadata: ObjectMetadata, *, cancel_check: Callable[[], None] | None = None) -> None:
+        self.verify_path(self.path_for(metadata.logical_sha256), metadata, cancel_check=cancel_check)
 
     @classmethod
-    def verify_path(cls, path: Path, metadata: ObjectMetadata) -> None:
-        actual = cls._inspect_path(path)
+    def verify_path(
+        cls, path: Path, metadata: ObjectMetadata, *, cancel_check: Callable[[], None] | None = None
+    ) -> None:
+        actual = cls._inspect_path(path, cancel_check=cancel_check)
         expected = (
             metadata.logical_sha256,
             metadata.compressed_sha256,
@@ -239,7 +243,7 @@ def verify_path(cls, path: Path, metadata: ObjectMetadata) -> None:
             raise ObjectStoreError(f"object verification failed for {path}")
 
     @staticmethod
-    def _inspect_path(path: Path) -> tuple[str, str, int, int]:
+    def _inspect_path(path: Path, *, cancel_check: Callable[[], None] | None = None) -> tuple[str, str, int, int]:
         compressed_hasher = hashlib.sha256()
         logical_hasher = hashlib.sha256()
         compressed_bytes = 0
@@ -247,6 +251,8 @@ def _inspect_path(path: Path) -> tuple[str, str, int, int]:
 
         with path.open("rb") as source:
             while True:
+                if cancel_check is not None:
+                    cancel_check()
                 chunk = source.read(1024 * 1024)
                 if not chunk:
                     break
@@ -257,6 +263,8 @@ def _inspect_path(path: Path) -> tuple[str, str, int, int]:
             reader = zstd.ZstdDecompressor().stream_reader(source)
             try:
                 while True:
+                    if cancel_check is not None:
+                        cancel_check()
                     chunk = reader.read(1024 * 1024)
                     if not chunk:
                         break
diff --git a/src/endstone_endkeep/repository/reader.py b/src/endstone_endkeep/repository/reader.py
index 662570e3..b0614dda 100644
--- a/src/endstone_endkeep/repository/reader.py
+++ b/src/endstone_endkeep/repository/reader.py
@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from contextlib import ExitStack
 
 from endstone_endkeep.logical.format import iter_base, iter_delta
@@ -21,6 +21,7 @@ def iter_state(
         manifest: RepositoryManifest,
         *,
         snapshot: str | None = None,
+        cancel_check: Callable[[], None] | None = None,
     ) -> Iterator[tuple[bytes, bytes]]:
         if snapshot is None:
             target_index = len(manifest.chain) - 1
@@ -34,7 +35,7 @@ def iter_state(
 
             for node in manifest.chain[1 : target_index + 1]:
                 delta_stream = stack.enter_context(self.objects.open_logical(node.object))
-                state = apply_delta(state, iter_delta(delta_stream))
+                state = apply_delta(state, iter_delta(delta_stream), cancel_check=cancel_check)
 
             yield from state
 
diff --git a/src/endstone_endkeep/repository/rollover.py b/src/endstone_endkeep/repository/rollover.py
index fb7f3225..71a217e4 100644
--- a/src/endstone_endkeep/repository/rollover.py
+++ b/src/endstone_endkeep/repository/rollover.py
@@ -64,13 +64,14 @@ def advance(amount: int) -> None:
             if cancel_check is not None:
                 cancel_check()
 
-        state = self.reader.iter_state(manifest, snapshot=target.snapshot)
+        state = self.reader.iter_state(manifest, snapshot=target.snapshot, cancel_check=cancel_check)
         object_meta, stats = self.objects.create(
             lambda stream: write_base(
                 stream,
                 state,
                 progress=advance if progress is not None or cancel_check is not None else None,
-            )
+            ),
+            cancel_check=cancel_check,
         )
         assert isinstance(stats, BaseStats)
 
diff --git a/src/endstone_endkeep/repository/verify.py b/src/endstone_endkeep/repository/verify.py
index 0c1a3d44..4d1cc63d 100644
--- a/src/endstone_endkeep/repository/verify.py
+++ b/src/endstone_endkeep/repository/verify.py
@@ -39,6 +39,7 @@ def verify(
         deep: bool = False,
         object_progress: ObjectProgressCallback | None = None,
         state_progress: StateProgressCallback | None = None,
+        cancel_check: Callable[[], None] | None = None,
     ) -> VerifyReport:
         manifest = self.manifests.load_current()
         if manifest is None:
@@ -54,6 +55,7 @@ def verify(
                 manifest,
                 object_progress=object_progress,
                 state_progress=state_progress,
+                cancel_check=cancel_check,
             )
 
         return VerifyReport(
@@ -88,6 +90,7 @@ def _verify_deep(
         *,
         object_progress: ObjectProgressCallback | None,
         state_progress: StateProgressCallback | None,
+        cancel_check: Callable[[], None] | None,
     ) -> None:
         checked: set[str] = set()
         for node in manifest.chain:
@@ -97,7 +100,7 @@ def _verify_deep(
 
                 if object_progress is not None:
                     object_progress(node, role, metadata, 0, 1)
-                self.objects.verify(metadata)
+                self.objects.verify(metadata, cancel_check=cancel_check)
                 if object_progress is not None:
                     object_progress(node, role, metadata, 1, 1)
                 checked.add(metadata.logical_sha256)
@@ -112,6 +115,8 @@ def advance_state(
                 current_node: SnapshotNode = node,
             ) -> None:
                 nonlocal completed_records
+                if cancel_check is not None:
+                    cancel_check()
                 completed_records += amount
                 if state_progress is not None:
                     state_progress(
@@ -121,8 +126,8 @@ def advance_state(
                     )
 
             stats = hash_state(
-                self.reader.iter_state(manifest, snapshot=node.snapshot),
-                progress=advance_state if state_progress is not None else None,
+                self.reader.iter_state(manifest, snapshot=node.snapshot, cancel_check=cancel_check),
+                progress=advance_state if state_progress is not None or cancel_check is not None else None,
             )
             if (
                 stats.sha256 != node.state_sha256
diff --git a/src/endstone_endkeep/staging/clone.py b/src/endstone_endkeep/staging/clone.py
index e5bc4ec7..27401fc9 100644
--- a/src/endstone_endkeep/staging/clone.py
+++ b/src/endstone_endkeep/staging/clone.py
@@ -3,8 +3,11 @@
 import os
 import shutil
 import stat
+from collections.abc import Callable
 from pathlib import Path
 
+from endstone_endkeep.repository.progress import JobCancelled
+
 _FICLONE = 0x40049409
 
 
@@ -12,7 +15,7 @@ class CloneError(RuntimeError):
     pass
 
 
-def _copy_reflink_or_bytes(source: Path, destination: Path) -> None:
+def _copy_reflink_or_bytes(source: Path, destination: Path, cancel_check: Callable[[], None] | None = None) -> None:
     source_fd = os.open(source, os.O_RDONLY | getattr(os, "O_BINARY", 0))
     destination_fd = os.open(destination, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0), 0o600)
     try:
@@ -28,6 +31,8 @@ def _copy_reflink_or_bytes(source: Path, destination: Path) -> None:
 
         if not cloned:
             while True:
+                if cancel_check is not None:
+                    cancel_check()
                 chunk = os.read(source_fd, 1024 * 1024)
                 if not chunk:
                     break
@@ -47,6 +52,8 @@ def clone_world(
     world_name: str,
     work_root: Path,
     snapshot_id: str,
+    *,
+    cancel_check: Callable[[], None] | None = None,
 ) -> Path:
     source = raw_snapshot_root / world_name
     if not source.is_dir():
@@ -60,6 +67,8 @@ def clone_world(
 
     try:
         for current, dirnames, filenames in os.walk(source, followlinks=False):
+            if cancel_check is not None:
+                cancel_check()
             current_path = Path(current)
             relative = current_path.relative_to(source)
             destination_dir = destination / relative
@@ -72,12 +81,16 @@ def clone_world(
                     raise CloneError(f"raw snapshot contains directory symlink: {child}")
 
             for filename in filenames:
+                if cancel_check is not None:
+                    cancel_check()
                 source_file = current_path / filename
                 info = source_file.lstat()
                 if not stat.S_ISREG(info.st_mode):
                     raise CloneError(f"raw snapshot contains non-regular file: {source_file}")
-                _copy_reflink_or_bytes(source_file, destination_dir / filename)
+                _copy_reflink_or_bytes(source_file, destination_dir / filename, cancel_check)
         return destination
+    except JobCancelled:
+        raise
     except Exception:
         shutil.rmtree(job_root, ignore_errors=True)
         raise

From 475877afc03df19d8f2ddfcfa560f4a42f333651 Mon Sep 17 00:00:00 2001
From: ReallocAll <ReallocAll@outlook.com>
Date: Thu, 15 Oct 2026 10:27:25 +0800
Subject: [PATCH 3/4] build: pin Amulet 3.0.8a1 and validate Python
 compatibility

---
 .github/workflows/compatibility.yml | 197 ++++++++--------------------
 pyproject.toml                      |  25 ++--
 requirements-offline.txt            |  21 ++-
 tests/conftest.py                   |   7 +
 tests/test_amulet_reader.py         |   9 ++
 5 files changed, 101 insertions(+), 158 deletions(-)
 create mode 100644 tests/conftest.py

diff --git a/.github/workflows/compatibility.yml b/.github/workflows/compatibility.yml
index 9f695927..8d0953c3 100644
--- a/.github/workflows/compatibility.yml
+++ b/.github/workflows/compatibility.yml
@@ -21,176 +21,83 @@ jobs:
       fail-fast: false
       matrix:
         include:
-          - { os: ubuntu-24.04, python: '3.12', wheel: cp312-manylinux_x86_64 }
-          - { os: ubuntu-24.04, python: '3.13', wheel: cp313-manylinux_x86_64 }
-          - { os: ubuntu-24.04, python: '3.14', wheel: '' }
-          - { os: ubuntu-24.04, python: '3.15', wheel: cp315-manylinux_x86_64 }
-          - { os: windows-2025, python: '3.12', wheel: cp312-win_amd64 }
-          - { os: windows-2025, python: '3.13', wheel: cp313-win_amd64 }
-          - { os: windows-2025, python: '3.14', wheel: '' }
-          - { os: windows-2025, python: '3.15', wheel: cp315-win_amd64 }
+          - { os: ubuntu-24.04, python: '3.12' }
+          - { os: ubuntu-24.04, python: '3.13' }
+          - { os: ubuntu-24.04, python: '3.14' }
+          - { os: ubuntu-24.04, python: '3.15' }
+          - { os: windows-2025, python: '3.12' }
+          - { os: windows-2025, python: '3.13' }
+          - { os: windows-2025, python: '3.14' }
+          - { os: windows-2025, python: '3.15' }
     defaults:
       run:
         shell: bash
     steps:
-      - name: Checkout EndKeep
+      - name: Checkout exact EndKeep source
         uses: actions/checkout@v7
         with:
           fetch-depth: 0
           persist-credentials: false
-      - name: Checkout upstream Amulet 3.0.7a0 at verified commit
-        if: matrix.python != '3.14'
-        uses: actions/checkout@v7
-        with:
-          repository: Amulet-Team/Amulet-LevelDB
-          ref: 736228e375e9f535b6522bd28168fc306a5354b9
-          path: amulet-src
-          fetch-depth: 0
-          persist-credentials: false
-      - name: Patch Amulet Python floor
-        if: matrix.python != '3.14'
-        run: |
-          python - <<'PY'
-          from pathlib import Path
-          project = Path("amulet-src/pyproject.toml")
-          content = project.read_text()
-          needle = 'requires-python = ">=3.14"'
-          assert content.count(needle) == 1
-          project.write_text(content.replace(needle, 'requires-python = ">=3.12"'))
 
-          PY
-          cp .github/scripts/amulet_leveldb_smoke.py amulet-src/tests/test_backport_smoke.py
-          git -C amulet-src diff --check
-      - name: Setup host Python
-        if: matrix.python != '3.15'
-        uses: actions/setup-python@v7
-        with:
-          python-version: ${{ matrix.python }}
-      - name: Install cibuildwheel
-        if: matrix.python != '3.14'
-        run: python -m pip install 'cibuildwheel==4.2.0'
-      - name: Build and test Amulet wheel
-        if: matrix.python != '3.14'
-        env:
-          CIBW_BEFORE_BUILD: 'python -m pip install "cmake>=4.1,<5"'
-          CIBW_BEFORE_TEST: 'python -m pip install "cmake>=4.1,<5"'
-          CIBW_TEST_COMMAND: 'python {project}/tests/test_backport_smoke.py && python {project}/tests/test_db.py'
-          CIBW_TEST_SOURCES: tests
-        run: |
-          mkdir -p amulet-dist
-          cd amulet-src
-          python -m cibuildwheel --only '${{ matrix.wheel }}' --output-dir ../amulet-dist .
-      - name: Setup uv
+      - name: Set up uv
         uses: astral-sh/setup-uv@v7
         with:
           enable-cache: true
-      - name: Download exact Endstone CPython 3.15 Wheel
-        if: matrix.python == '3.15'
-        uses: actions/download-artifact@v8
-        with:
-          repository: EndstoneMC/endstone
-          run-id: 38045137884
-          github-token: ${{ github.token }}
-          pattern: endstone-*-cp315-cp315-*.whl
-          path: endstone-dist
-          merge-multiple: true
-          skip-decompress: true
 
-      - name: Test installed EndKeep with matched Amulet
+      - name: Install selected interpreter
         run: |
+          uv python install '${{ matrix.python }}'
           uv venv .venv --python '${{ matrix.python }}'
           if [[ "${{ runner.os }}" == "Windows" ]]; then
-            PYTHON=.venv/Scripts/python.exe
+            echo 'PYTHON=.venv/Scripts/python.exe' >> "$GITHUB_ENV"
           else
-            PYTHON=.venv/bin/python
+            echo 'PYTHON=.venv/bin/python' >> "$GITHUB_ENV"
           fi
+
+      - name: Download compatible experimental Endstone for Python 3.15
+        if: matrix.python == '3.15'
+        env:
+          GH_TOKEN: ${{ secrets.REPO_PAT || github.token }}
+        run: |
+          if [[ "${{ runner.os }}" == "Windows" ]]; then
+            WHEEL='endstone-0.11.14.dev6-cp315-cp315-win_amd64.whl'
+          else
+            WHEEL='endstone-0.11.14.dev6-cp315-cp315-manylinux_2_31_x86_64.whl'
+          fi
+          mkdir -p endstone-artifact
+          gh run download 38045137884 --repo EndstoneMC/endstone --name "$WHEEL" --dir endstone-artifact
+          ENDSTONE_WHEEL="$WHEEL" python - <<'PY'
+          import os
+          from pathlib import Path
+          from zipfile import ZIP_DEFLATED, ZipFile
+          root = Path("endstone-artifact")
+          target = root / os.environ["ENDSTONE_WHEEL"]
+          members = sorted(p for p in root.rglob("*") if p.is_file() and p != target)
+          assert any(".dist-info/" in p.as_posix() for p in members)
+          with ZipFile(target, "w", compression=ZIP_DEFLATED, compresslevel=1) as wheel:
+              for path in members:
+                  wheel.write(path, path.relative_to(root).as_posix())
+          with ZipFile(target) as wheel:
+              assert wheel.testzip() is None
+          PY
+          echo "ENDSTONE_WHEEL=endstone-artifact/$WHEEL" >> "$GITHUB_ENV"
+
+      - name: Install pinned compatibility dependencies
+        run: |
           if [[ "${{ matrix.python }}" == "3.15" ]]; then
-            if [[ "${{ runner.os }}" == "Windows" ]]; then
-              ENDSTONE_WHEEL="$(find endstone-dist -name '*cp315*win_amd64.whl' -print -quit)"
-            else
-              ENDSTONE_WHEEL="$(find endstone-dist -name '*cp315*manylinux*.whl' -print -quit)"
-            fi
-            test -n "$ENDSTONE_WHEEL"
             uv pip install --python "$PYTHON" "$ENDSTONE_WHEEL"
           fi
-          # Fresh dependency resolution: do NOT seed the local Amulet build.
-          # This must fetch the exact published GitHub Release Wheel on 3.12/3.13/3.15.
           uv pip install --python "$PYTHON" -e '.[dev]'
-          "$PYTHON" - <<'PY'
-          from importlib.metadata import distribution
-          from packaging.requirements import Requirement
-          requires = distribution("endstone-endkeep").requires or []
-          amulet = [Requirement(req) for req in requires if Requirement(req).name == "amulet-leveldb"]
-          assert len(amulet) == 7, [str(req) for req in amulet]
-          print("Packaged Amulet selectors:", *amulet, sep="\\n")
-          PY
           uv pip check --python "$PYTHON"
           "$PYTHON" - <<'PY'
           from importlib.metadata import version
           from amulet.leveldb import LevelDB
-          assert version("amulet-leveldb").split("+")[0] == "3.0.7a0"
-          print("Amulet:", version("amulet-leveldb"))
+          assert version("amulet-leveldb").split("+")[0] == "3.0.8a1"
+          assert version("amulet-utils").split("+")[0] == "1.1.8a0"
+          print("Amulet-LevelDB:", version("amulet-leveldb"))
+          print("Amulet-Utils:", version("amulet-utils"))
+          print("Endstone:", version("endstone"))
           PY
-          "$PYTHON" -m pytest -q
-      - name: Save tested experimental backports (no Release)
-        if: matrix.python != '3.14'
-        uses: actions/upload-artifact@v7
-        with:
-          name: amulet-compat-${{ matrix.os }}-${{ matrix.python }}
-          path: amulet-dist/*.whl
-          if-no-files-found: error
-          retention-days: 7
 
-  verified-backports:
-    name: Collect verified compatibility Wheels
-    needs: full-matrix
-    runs-on: ubuntu-24.04
-    steps:
-      - name: Download tested wheel artifacts
-        uses: actions/download-artifact@v8
-        with:
-          pattern: amulet-compat-*
-          path: dist
-          merge-multiple: true
-
-      - name: Validate Wheels and create SHA256 manifest
-        run: |
-          python - <<'PY'
-          from hashlib import sha256
-          from pathlib import Path
-          from zipfile import ZipFile
-
-          root = Path("dist")
-          wheels = sorted(root.glob("*.whl"))
-          assert len(wheels) == 6, [p.name for p in wheels]
-          for abi in ("cp312", "cp313", "cp315"):
-              for platform in ("manylinux_", "win_amd64"):
-                  matches = [
-                      p for p in wheels
-                      if f"-{abi}-{abi}-" in p.name and platform in p.name
-                  ]
-                  assert len(matches) == 1, (abi, platform, [p.name for p in wheels])
-          for wheel in wheels:
-              with ZipFile(wheel) as archive:
-                  meta = next(path for path in archive.namelist() if path.endswith(".dist-info/METADATA"))
-                  data = archive.read(meta).decode("utf-8")
-                  assert "Name: amulet-leveldb" in data
-                  assert "Version: 3.0.7a0+0.g736228e.dirty" in data
-                  assert "Requires-Python: >=3.12" in data
-                  assert any(path.endswith(".dist-info/licenses/LICENSE") for path in archive.namelist())
-          (root / "SHA256SUMS").write_text("".join(
-              f"{sha256(path.read_bytes()).hexdigest()}  {path.name}\n"
-              for path in wheels
-          ))
-          print((root / "SHA256SUMS").read_text())
-          PY
-
-      - name: Upload verified wheel set for review (not a GitHub Release)
-        uses: actions/upload-artifact@v7
-        with:
-          name: amulet-backports-verified-six-wheels
-          path: |
-            dist/*.whl
-            dist/SHA256SUMS
-          if-no-files-found: error
-          retention-days: 14
+      - name: Run regression suite
+        run: '"$PYTHON" -m pytest -q'
diff --git a/pyproject.toml b/pyproject.toml
index 5e16e41d..cb9dd6ef 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -7,14 +7,23 @@ name = "endstone-endkeep"
 dynamic = ["version"]
 dependencies = [
     "endstone>=0.11,<0.12",
-    # Official upstream Wheel for Python 3.14; SHA256-pinned compatibility Wheels for 3.12, 3.13 and 3.15.
-    "amulet-leveldb==3.0.7a0; python_version == '3.14'",
-    'amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.7a0-compat-cp312-cp313/amulet_leveldb-3.0.7a0%2B0.g736228e.dirty-cp312-cp312-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl#sha256=283e7e57e4d15d7ee59457a89db8c43be9a1cd85c7174844f3a1f1bd14d93bbd ; python_version == "3.12" and sys_platform == "linux" and platform_machine == "x86_64"',
-    'amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.7a0-compat-cp312-cp313/amulet_leveldb-3.0.7a0%2B0.g736228e.dirty-cp312-cp312-win_amd64.whl#sha256=9033156a570ceef3613486a681b990f442788b636da1018838eb6ffdd3dca73f ; python_version == "3.12" and sys_platform == "win32" and platform_machine == "AMD64"',
-    'amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.7a0-compat-cp312-cp313/amulet_leveldb-3.0.7a0%2B0.g736228e.dirty-cp313-cp313-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl#sha256=718d5ff3887b4d848f811a49ab6a7fe3e0b1f863821029190714c4ea5362d9fc ; python_version == "3.13" and sys_platform == "linux" and platform_machine == "x86_64"',
-    'amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.7a0-compat-cp312-cp313/amulet_leveldb-3.0.7a0%2B0.g736228e.dirty-cp313-cp313-win_amd64.whl#sha256=28dc6037e3d16355a5120baba1ec61065d020f65500ca9de20d97deaaac0ab2f ; python_version == "3.13" and sys_platform == "win32" and platform_machine == "AMD64"',
-    'amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.7a0-compat-cp315/amulet_leveldb-3.0.7a0%2B0.g736228e.dirty-cp315-cp315-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl#sha256=7fec1e03a2d72d1c08944726c08909fd811cf7ad905e6f124cea581bcfc554ad ; python_version == "3.15" and sys_platform == "linux" and platform_machine == "x86_64"',
-    'amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.7a0-compat-cp315/amulet_leveldb-3.0.7a0%2B0.g736228e.dirty-cp315-cp315-win_amd64.whl#sha256=a54e2004f50b8fe5ac2a23edb7cd5c5e60fbc9a8d44fadf282e17b0d3416a611 ; python_version == "3.15" and sys_platform == "win32" and platform_machine == "AMD64"',
+    # SHA256-verified compatibility wheels from the maintained 3.0.8a1 release.
+    'amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_leveldb-3.0.8a1%2B0.g18e624b.dirty-cp312-cp312-manylinux_2_34_x86_64.whl#sha256=6e18394f78fc350851dbf049205c15f4e9fa6050fd8381e1ea3635a76510e742 ; python_version == "3.12" and sys_platform == "linux" and platform_machine == "x86_64"',
+    'amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_leveldb-3.0.8a1%2B0.g18e624b.dirty-cp312-cp312-win_amd64.whl#sha256=ba96ed2d9e614021b6d32feadc7aee8858318bfc8b96c13d474e729c4b63117d ; python_version == "3.12" and sys_platform == "win32" and platform_machine == "AMD64"',
+    'amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_leveldb-3.0.8a1%2B0.g18e624b.dirty-cp313-cp313-manylinux_2_34_x86_64.whl#sha256=3825aa7baf86aae33dff6541b99fe5757e00c176e44edfc5576f172bdbd1e9aa ; python_version == "3.13" and sys_platform == "linux" and platform_machine == "x86_64"',
+    'amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_leveldb-3.0.8a1%2B0.g18e624b.dirty-cp313-cp313-win_amd64.whl#sha256=54a0bbda4e8405e4836a0a012915ad54b8794a40bd0865f54aaab3906553e348 ; python_version == "3.13" and sys_platform == "win32" and platform_machine == "AMD64"',
+    'amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_leveldb-3.0.8a1%2B0.g18e624b.dirty-cp314-cp314-manylinux_2_34_x86_64.whl#sha256=250d4f00cf891a36882b9b9545e88998ea9193db8bac57f96424558740ada738 ; python_version == "3.14" and sys_platform == "linux" and platform_machine == "x86_64"',
+    'amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_leveldb-3.0.8a1%2B0.g18e624b.dirty-cp314-cp314-win_amd64.whl#sha256=dff3c39391be7fad659d82cd452ce1bfaf964a68060c8318016293bfb08cc093 ; python_version == "3.14" and sys_platform == "win32" and platform_machine == "AMD64"',
+    'amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_leveldb-3.0.8a1%2B0.g18e624b.dirty-cp315-cp315-manylinux_2_34_x86_64.whl#sha256=d6eb76a49ab0c2a8e76f5078a4d9f70a40e6821b80f600d4344f24293973f950 ; python_version == "3.15" and sys_platform == "linux" and platform_machine == "x86_64"',
+    'amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_leveldb-3.0.8a1%2B0.g18e624b.dirty-cp315-cp315-win_amd64.whl#sha256=8e9fc3a16aa9cef97482512e20a1efdf9fae41a4f042efdc7671930e229521db ; python_version == "3.15" and sys_platform == "win32" and platform_machine == "AMD64"',
+    'amulet-utils @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_utils-1.1.8a0%2B0.g1655a56.dirty-cp312-cp312-manylinux_2_34_x86_64.whl#sha256=072bf7ccffc4bc0bf1ca9cd522d9cf73c80b665b931f1e18d409346745fd0b13 ; python_version == "3.12" and sys_platform == "linux" and platform_machine == "x86_64"',
+    'amulet-utils @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_utils-1.1.8a0%2B0.g1655a56.dirty-cp312-cp312-win_amd64.whl#sha256=b1f0512e7f2b061eb1cfd61f5aa10e8704c1378bf2d77fb7282543a49d4a839b ; python_version == "3.12" and sys_platform == "win32" and platform_machine == "AMD64"',
+    'amulet-utils @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_utils-1.1.8a0%2B0.g1655a56.dirty-cp313-cp313-manylinux_2_34_x86_64.whl#sha256=c6401a8a6b33c1a01adafb5632633fdda54315ebdaca44cd7a17d0d7a5d20377 ; python_version == "3.13" and sys_platform == "linux" and platform_machine == "x86_64"',
+    'amulet-utils @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_utils-1.1.8a0%2B0.g1655a56.dirty-cp313-cp313-win_amd64.whl#sha256=8dbf688ba8ac46a816fab85fc18f2fe8aa217af73aaa8fe4d23b2b0053254cfa ; python_version == "3.13" and sys_platform == "win32" and platform_machine == "AMD64"',
+    'amulet-utils @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_utils-1.1.8a0%2B0.g1655a56.dirty-cp314-cp314-manylinux_2_34_x86_64.whl#sha256=ce869069252adb5d542bc62ada283d8a8a99b1f27a193cab19c239435a7026ae ; python_version == "3.14" and sys_platform == "linux" and platform_machine == "x86_64"',
+    'amulet-utils @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_utils-1.1.8a0%2B0.g1655a56.dirty-cp314-cp314-win_amd64.whl#sha256=558a4e57624fcd12e0c68d7dc4c70fd514c112f10cad130f171e5764a0082003 ; python_version == "3.14" and sys_platform == "win32" and platform_machine == "AMD64"',
+    'amulet-utils @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_utils-1.1.8a0%2B0.g1655a56.dirty-cp315-cp315-manylinux_2_34_x86_64.whl#sha256=3eccf5d0ad1b84a32bdcfadf5e168db4fd567730f3cf5ee7fea597168f50a1e8 ; python_version == "3.15" and sys_platform == "linux" and platform_machine == "x86_64"',
+    'amulet-utils @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_utils-1.1.8a0%2B0.g1655a56.dirty-cp315-cp315-win_amd64.whl#sha256=d52afdbf279c982227a8eab69662080615b36a936d13a3223073a6be0149ece5 ; python_version == "3.15" and sys_platform == "win32" and platform_machine == "AMD64"',
     "zstandard>=0.23,<1",
     "tomlkit>=0.13,<1",
 ]
diff --git a/requirements-offline.txt b/requirements-offline.txt
index f96e7629..60a50103 100644
--- a/requirements-offline.txt
+++ b/requirements-offline.txt
@@ -1,7 +1,18 @@
-amulet-leveldb==3.0.7a0 ; python_version == '3.14'
-amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.7a0-compat-cp312-cp313/amulet_leveldb-3.0.7a0%2B0.g736228e.dirty-cp312-cp312-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl#sha256=283e7e57e4d15d7ee59457a89db8c43be9a1cd85c7174844f3a1f1bd14d93bbd ; python_version == "3.12" and sys_platform == "linux" and platform_machine == "x86_64"
-amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.7a0-compat-cp312-cp313/amulet_leveldb-3.0.7a0%2B0.g736228e.dirty-cp312-cp312-win_amd64.whl#sha256=9033156a570ceef3613486a681b990f442788b636da1018838eb6ffdd3dca73f ; python_version == "3.12" and sys_platform == "win32" and platform_machine == "AMD64"
-amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.7a0-compat-cp312-cp313/amulet_leveldb-3.0.7a0%2B0.g736228e.dirty-cp313-cp313-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl#sha256=718d5ff3887b4d848f811a49ab6a7fe3e0b1f863821029190714c4ea5362d9fc ; python_version == "3.13" and sys_platform == "linux" and platform_machine == "x86_64"
-amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.7a0-compat-cp312-cp313/amulet_leveldb-3.0.7a0%2B0.g736228e.dirty-cp313-cp313-win_amd64.whl#sha256=28dc6037e3d16355a5120baba1ec61065d020f65500ca9de20d97deaaac0ab2f ; python_version == "3.13" and sys_platform == "win32" and platform_machine == "AMD64"
+amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_leveldb-3.0.8a1%2B0.g18e624b.dirty-cp312-cp312-manylinux_2_34_x86_64.whl#sha256=6e18394f78fc350851dbf049205c15f4e9fa6050fd8381e1ea3635a76510e742 ; python_version == "3.12" and sys_platform == "linux" and platform_machine == "x86_64"
+amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_leveldb-3.0.8a1%2B0.g18e624b.dirty-cp312-cp312-win_amd64.whl#sha256=ba96ed2d9e614021b6d32feadc7aee8858318bfc8b96c13d474e729c4b63117d ; python_version == "3.12" and sys_platform == "win32" and platform_machine == "AMD64"
+amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_leveldb-3.0.8a1%2B0.g18e624b.dirty-cp313-cp313-manylinux_2_34_x86_64.whl#sha256=3825aa7baf86aae33dff6541b99fe5757e00c176e44edfc5576f172bdbd1e9aa ; python_version == "3.13" and sys_platform == "linux" and platform_machine == "x86_64"
+amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_leveldb-3.0.8a1%2B0.g18e624b.dirty-cp313-cp313-win_amd64.whl#sha256=54a0bbda4e8405e4836a0a012915ad54b8794a40bd0865f54aaab3906553e348 ; python_version == "3.13" and sys_platform == "win32" and platform_machine == "AMD64"
+amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_leveldb-3.0.8a1%2B0.g18e624b.dirty-cp314-cp314-manylinux_2_34_x86_64.whl#sha256=250d4f00cf891a36882b9b9545e88998ea9193db8bac57f96424558740ada738 ; python_version == "3.14" and sys_platform == "linux" and platform_machine == "x86_64"
+amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_leveldb-3.0.8a1%2B0.g18e624b.dirty-cp314-cp314-win_amd64.whl#sha256=dff3c39391be7fad659d82cd452ce1bfaf964a68060c8318016293bfb08cc093 ; python_version == "3.14" and sys_platform == "win32" and platform_machine == "AMD64"
+amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_leveldb-3.0.8a1%2B0.g18e624b.dirty-cp315-cp315-manylinux_2_34_x86_64.whl#sha256=d6eb76a49ab0c2a8e76f5078a4d9f70a40e6821b80f600d4344f24293973f950 ; python_version == "3.15" and sys_platform == "linux" and platform_machine == "x86_64"
+amulet-leveldb @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_leveldb-3.0.8a1%2B0.g18e624b.dirty-cp315-cp315-win_amd64.whl#sha256=8e9fc3a16aa9cef97482512e20a1efdf9fae41a4f042efdc7671930e229521db ; python_version == "3.15" and sys_platform == "win32" and platform_machine == "AMD64"
+amulet-utils @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_utils-1.1.8a0%2B0.g1655a56.dirty-cp312-cp312-manylinux_2_34_x86_64.whl#sha256=072bf7ccffc4bc0bf1ca9cd522d9cf73c80b665b931f1e18d409346745fd0b13 ; python_version == "3.12" and sys_platform == "linux" and platform_machine == "x86_64"
+amulet-utils @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_utils-1.1.8a0%2B0.g1655a56.dirty-cp312-cp312-win_amd64.whl#sha256=b1f0512e7f2b061eb1cfd61f5aa10e8704c1378bf2d77fb7282543a49d4a839b ; python_version == "3.12" and sys_platform == "win32" and platform_machine == "AMD64"
+amulet-utils @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_utils-1.1.8a0%2B0.g1655a56.dirty-cp313-cp313-manylinux_2_34_x86_64.whl#sha256=c6401a8a6b33c1a01adafb5632633fdda54315ebdaca44cd7a17d0d7a5d20377 ; python_version == "3.13" and sys_platform == "linux" and platform_machine == "x86_64"
+amulet-utils @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_utils-1.1.8a0%2B0.g1655a56.dirty-cp313-cp313-win_amd64.whl#sha256=8dbf688ba8ac46a816fab85fc18f2fe8aa217af73aaa8fe4d23b2b0053254cfa ; python_version == "3.13" and sys_platform == "win32" and platform_machine == "AMD64"
+amulet-utils @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_utils-1.1.8a0%2B0.g1655a56.dirty-cp314-cp314-manylinux_2_34_x86_64.whl#sha256=ce869069252adb5d542bc62ada283d8a8a99b1f27a193cab19c239435a7026ae ; python_version == "3.14" and sys_platform == "linux" and platform_machine == "x86_64"
+amulet-utils @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_utils-1.1.8a0%2B0.g1655a56.dirty-cp314-cp314-win_amd64.whl#sha256=558a4e57624fcd12e0c68d7dc4c70fd514c112f10cad130f171e5764a0082003 ; python_version == "3.14" and sys_platform == "win32" and platform_machine == "AMD64"
+amulet-utils @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_utils-1.1.8a0%2B0.g1655a56.dirty-cp315-cp315-manylinux_2_34_x86_64.whl#sha256=3eccf5d0ad1b84a32bdcfadf5e168db4fd567730f3cf5ee7fea597168f50a1e8 ; python_version == "3.15" and sys_platform == "linux" and platform_machine == "x86_64"
+amulet-utils @ https://github.com/ReallocAll/endkeep/releases/download/amulet-leveldb-3.0.8a1-compat-cp312-cp315/amulet_utils-1.1.8a0%2B0.g1655a56.dirty-cp315-cp315-win_amd64.whl#sha256=d52afdbf279c982227a8eab69662080615b36a936d13a3223073a6be0149ece5 ; python_version == "3.15" and sys_platform == "win32" and platform_machine == "AMD64"
 zstandard>=0.23,<1
 tqdm>=4.70.1,<5
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 00000000..015d5d3a
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,7 @@
+import ctypes
+import sys
+
+if sys.platform == "linux":
+    # Tests share a process; production keeps Amulet in the separate Worker.
+    # Load GCC's unwinder before Endstone's bundled LLVM libunwind.
+    ctypes.CDLL("libgcc_s.so.1", mode=ctypes.RTLD_GLOBAL)
diff --git a/tests/test_amulet_reader.py b/tests/test_amulet_reader.py
index ed54f339..7b52e91e 100644
--- a/tests/test_amulet_reader.py
+++ b/tests/test_amulet_reader.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from concurrent.futures import ThreadPoolExecutor
+from importlib import import_module
 from pathlib import Path
 
 import pytest
@@ -40,3 +41,11 @@ def test_iter_visible_state_exhausts_in_repository_worker_thread(tmp_path: Path)
         result = executor.submit(_read_all, db_path).result(timeout=10)
 
     assert result == expected
+
+
+def test_missing_leveldb_raises_python_exception(tmp_path: Path) -> None:
+    import_module("endstone._python")
+    from amulet.leveldb import LevelDB, LevelDBException
+
+    with pytest.raises(LevelDBException, match="No database exists"):
+        LevelDB(str(tmp_path / "missing"), False)

From 0672f0847001f5ff74bce39a273dc796e6eaad95 Mon Sep 17 00:00:00 2001
From: ReallocAll <ReallocAll@outlook.com>
Date: Thu, 15 Oct 2026 10:27:30 +0800
Subject: [PATCH 4/4] test: cover worker shutdown, recovery and supported
 installation paths

---
 README.md                          |  22 +-
 docs/manual-validation.md          |   3 +-
 docs/using-endkeep.md              |  25 +-
 tests/test_offline_mutation.py     |   4 +-
 tests/test_online_management.py    |   2 -
 tests/test_repository_health.py    |   8 +-
 tests/test_shutdown.py             | 312 +++++++++++++++++++++
 tests/test_worker.py               | 431 ++++++++++++++++++++++++++++-
 tests/transaction_crash_fixture.py |  37 +++
 9 files changed, 822 insertions(+), 22 deletions(-)
 create mode 100644 tests/test_shutdown.py
 create mode 100644 tests/transaction_crash_fixture.py

diff --git a/README.md b/README.md
index 6076818b..34b70fc0 100644
--- a/README.md
+++ b/README.md
@@ -10,9 +10,25 @@ without storing a complete world copy each time, and lets you restore an earlier
 
 Requires **Endstone 0.11** and **Python 3.12–3.15**.
 
-1. Download the plugin `.whl` from [Releases](https://github.com/ReallocAll/endkeep/releases/latest).
-2. Place it in the server's `plugins/` directory.
-3. Restart Endstone. Dependencies install automatically (internet access required).
+Download the plugin `.whl` from [Releases](https://github.com/ReallocAll/endkeep/releases/latest).
+Installation requires internet access.
+
+**Windows:** stop Endstone, activate the Python environment used to run it,
+and install the downloaded wheel (replace `VERSION` with its actual version):
+
+```sh
+python -m pip install --upgrade "./endstone_endkeep-VERSION-py3-none-any.whl"
+```
+
+Remove any EndKeep wheel from `plugins/`, then restart Endstone.
+Use the same steps for updates.
+
+**Linux:** use the command above, or place the wheel in `plugins/` and restart
+Endstone to install dependencies automatically.
+
+Python 3.15 currently requires compiler tools to build Zstandard.
+Before using the Linux `plugins/` method with Python 3.15, install it in the
+server's Python environment: `python -m pip install "zstandard>=0.23,<1"`.
 
 See [Using EndKeep](docs/using-endkeep.md) for commands and configuration.
 
diff --git a/docs/manual-validation.md b/docs/manual-validation.md
index 1cacc872..ec937e49 100644
--- a/docs/manual-validation.md
+++ b/docs/manual-validation.md
@@ -6,7 +6,8 @@ repository. These checks complement automated tests.
 ## Installation and backup
 
 1. Install the release wheel on supported platforms and Python versions.
-   Confirm the plugin and its dependencies load without manual setup.
+   Use command-line installation on Windows; check both installation methods on Linux.
+   Confirm the plugin and its dependencies load on unmodified Endstone.
 2. Capture multiple snapshots while BDS is running, process maintenance,
    and run `/backup verify deep`.
 3. Check backup scheduling, retention, low-disk-space protection and reload.
diff --git a/docs/using-endkeep.md b/docs/using-endkeep.md
index b773ba50..6bbea869 100644
--- a/docs/using-endkeep.md
+++ b/docs/using-endkeep.md
@@ -5,9 +5,8 @@ It stores snapshots in the directory configured by `storage.path` (default: `bac
 
 ## Getting started
 
-Download the plugin `.whl` from [Releases](https://github.com/ReallocAll/endkeep/releases/latest),
-place it in `plugins/` and restart Endstone. Dependencies install automatically
-on the first start (internet access required).
+Follow the [installation instructions](../README.md#installation): Windows uses
+command-line installation; Linux also supports placing the wheel in `plugins/`.
 
 Check `/backup status`, take a snapshot with `/backup create`, and run
 `/backup maintenance` to process it. View recovery points with `/backup list`.
@@ -34,8 +33,8 @@ Commands are available to the console and operators by default (`endkeep.admin`)
 
 Deleting or rolling over a snapshot is a two-step operation: run the command
 to view what will change, then run `/backup confirm`. Preview information is
-cleared on reload or restart. Export, delete and rollover jobs cannot be
-cancelled once accepted. Make an independent copy before changing recovery history.
+cleared on reload or restart. Delete and rollover jobs can be cancelled before the atomic commit; exports
+can also be cancelled. Make an independent copy before changing recovery history.
 
 ## Automatic backups
 
@@ -71,3 +70,19 @@ is running. To recover when the server cannot start, use the standalone restore
 tool. Neither method overwrites your active world.
 
 Follow the [Recovery guide](recovery.md) before replacing a live world.
+
+## Server shutdown and plugin reload
+
+On server shutdown, the worker cancels background work at a safe checkpoint and
+exits. An atomic commit already in progress finishes first; uncommitted RAW
+snapshots remain available for the next startup.
+This also applies after an unexpected server process exit. Unacknowledged
+results are recovered on the next startup.
+
+Plugin disable allows 15 seconds for a newly enabled instance to reconnect.
+Running repository jobs continue through the handoff. If the plugin stays
+unloaded, the worker safely cancels its job and exits.
+
+Startup briefly retries an unavailable worker. If it remains alive but cannot
+respond, EndKeep refuses a second writer. Check `backups/worker.log` and keep
+the backup directory intact.
diff --git a/tests/test_offline_mutation.py b/tests/test_offline_mutation.py
index ec59f4d0..507ed556 100644
--- a/tests/test_offline_mutation.py
+++ b/tests/test_offline_mutation.py
@@ -93,9 +93,9 @@ def test_middle_delete_replays_predecessor_once_and_cancels_transient_keys(tmp_p
     original_iter_state = RepositoryReader.iter_state
     replays: list[str | None] = []
 
-    def tracked_iter_state(self, manifest, *, snapshot=None):
+    def tracked_iter_state(self, manifest, *, snapshot=None, cancel_check=None):
         replays.append(snapshot)
-        return original_iter_state(self, manifest, snapshot=snapshot)
+        return original_iter_state(self, manifest, snapshot=snapshot, cancel_check=cancel_check)
 
     monkeypatch.setattr(RepositoryReader, "iter_state", tracked_iter_state)
     with RepositoryLock(store.repo_root):
diff --git a/tests/test_online_management.py b/tests/test_online_management.py
index af10168a..dbddfb76 100644
--- a/tests/test_online_management.py
+++ b/tests/test_online_management.py
@@ -60,7 +60,6 @@ def test_worker_mutation_preview_and_generation_guard(tmp_path: Path) -> None:
         assert plan["detail"] == f"replace BASE with {S2}"
         assert plan["remaining"] == 1
         assert service.start_mutation("delete", S1, plan["generation"])
-        assert service.request_cancel() is False
 
         # A repository task cannot be accepted while another job is running.
         assert (
@@ -149,7 +148,6 @@ def test_export_rejects_unsafe_and_existing_destinations(tmp_path: Path) -> None
                 service.start_export(invalid)
 
         assert service.start_export(S2)
-        assert service.request_cancel() is False
         assert _result(service).kind == "export"
         assert service.start_export(S2)
         with pytest.raises(FileExistsError, match="already exists"):
diff --git a/tests/test_repository_health.py b/tests/test_repository_health.py
index d55c1b0a..e1de15cb 100644
--- a/tests/test_repository_health.py
+++ b/tests/test_repository_health.py
@@ -44,10 +44,12 @@ def test_failed_deep_verify_persists_across_worker_restart(tmp_path: Path, monke
 
     original_verify = RepositoryVerifier.verify
 
-    def fail_verify(self, *, deep=False, object_progress=None, state_progress=None):
+    def fail_verify(self, *, deep=False, object_progress=None, state_progress=None, cancel_check=None):
         if deep:
             raise VerificationError("injected invalid state")
-        return original_verify(self, deep=deep, object_progress=object_progress, state_progress=state_progress)
+        return original_verify(
+            self, deep=deep, object_progress=object_progress, state_progress=state_progress, cancel_check=cancel_check
+        )
 
     service = _service(storage)
     try:
@@ -141,7 +143,7 @@ def test_full_verification_failure_blocks_next_maintenance(tmp_path: Path, monke
     build_fixture(storage)
     service = _service(storage)
 
-    def fail_verify(self, *, deep=False, object_progress=None, state_progress=None):
+    def fail_verify(self, *, deep=False, object_progress=None, state_progress=None, cancel_check=None):
         raise VerificationError("injected FULL verification failure")
 
     try:
diff --git a/tests/test_shutdown.py b/tests/test_shutdown.py
new file mode 100644
index 00000000..cdacc586
--- /dev/null
+++ b/tests/test_shutdown.py
@@ -0,0 +1,312 @@
+from __future__ import annotations
+
+import io
+import threading
+import time
+from pathlib import Path, PurePosixPath
+
+import pytest
+
+from endstone_endkeep.bds.model import SnapshotEntry
+from endstone_endkeep.logical.sidecar import write_sidecar
+from endstone_endkeep.repository.objects import ObjectStore
+from endstone_endkeep.repository.progress import JobCancelled
+from endstone_endkeep.repository.transaction import RepositoryTransaction
+from endstone_endkeep.staging.clone import clone_world
+from endstone_endkeep.worker.server import WorkerApplication
+from tests.standalone_fixture import add_raw
+from tests.test_worker import _settings
+
+
+def test_clone_cancel_preserves_raw_during_byte_copy(tmp_path: Path, monkeypatch) -> None:
+    raw = tmp_path / "raw"
+    source = raw / "level" / "db" / "large.ldb"
+    source.parent.mkdir(parents=True)
+    original = b"raw" * (1024 * 1024)
+    source.write_bytes(original)
+    work = tmp_path / "work"
+    destination = work / "s1" / "level" / "db" / "large.ldb"
+    monkeypatch.setattr("endstone_endkeep.staging.clone._FICLONE", 0)
+
+    def cancel() -> None:
+        if destination.exists() and destination.stat().st_size >= 1024 * 1024:
+            raise JobCancelled("stop")
+
+    with pytest.raises(JobCancelled):
+        clone_world(raw, "level", work, "s1", cancel_check=cancel)
+    assert source.read_bytes() == original
+    assert destination.exists(), "cancellation should defer clone cleanup until startup"
+
+
+def test_sidecar_cancel_preserves_raw(tmp_path: Path) -> None:
+    raw = tmp_path / "raw"
+    raw.mkdir()
+    original = b"sidecar" * (512 * 1024)
+    (raw / "large").write_bytes(original)
+    stream = io.BytesIO()
+
+    def cancel() -> None:
+        if stream.tell() >= 1024 * 1024:
+            raise JobCancelled("stop")
+
+    with pytest.raises(JobCancelled):
+        write_sidecar(stream, raw, [SnapshotEntry(PurePosixPath("large"), len(original))], cancel_check=cancel)
+    assert len(stream.getvalue()) < len(original)
+    assert (raw / "large").read_bytes() == original
+
+
+def test_object_verification_can_cancel_without_changing_object(tmp_path: Path) -> None:
+    objects = ObjectStore(tmp_path / "repo", compression_level=1, compression_threads=1)
+    metadata, _ = objects.create(lambda stream: stream.write(b"data" * (1024 * 1024)))
+    path = objects.path_for(metadata.logical_sha256)
+    original = path.read_bytes()
+
+    def cancel() -> None:
+        raise JobCancelled("stop")
+
+    with pytest.raises(JobCancelled):
+        objects.verify(metadata, cancel_check=cancel)
+    assert path.read_bytes() == original
+    objects.verify(metadata)
+
+
+def test_controller_death_finishes_commit_then_cancels_remaining_raw(tmp_path: Path, monkeypatch) -> None:
+    storage = tmp_path / "backups"
+    dead = threading.Event()
+    monkeypatch.setattr(
+        "endstone_endkeep.worker.server.process_identity",
+        lambda pid: "new" if pid == 200 else (None if dead.is_set() else "owner"),
+    )
+    app = WorkerApplication(storage, priority="background", controller_pid=100, controller_start="owner")
+    app.handle({"command": "configure", "settings": _settings()})
+    first, second = "20261014-120000", "20261014-130000"
+    add_raw(storage, first, [(b"a", b"first")], b"first")
+    add_raw(storage, second, [(b"a", b"second")], b"second")
+    entered, release, stopped = threading.Event(), threading.Event(), threading.Event()
+    commit = RepositoryTransaction.commit
+
+    def blocked_commit(transaction, manifest):
+        def hook(point):
+            if point == "after_manifest_fsync":
+                entered.set()
+                assert release.wait(5)
+
+        transaction.fault_hook = hook
+        return commit(transaction, manifest)
+
+    monkeypatch.setattr(RepositoryTransaction, "commit", blocked_commit)
+    app.set_stop_callback(stopped.set)
+    monitor = threading.Thread(target=app.monitor, daemon=True)
+    monitor.start()
+    try:
+        assert app.handle({"command": "start_maintenance", "mode": "LOGIC_ONLY"})["accepted"]
+        assert entered.wait(2)
+        dead.set()
+        deadline = time.monotonic() + 2
+        while not app.shutdown_requested and time.monotonic() < deadline:
+            time.sleep(0.02)
+        assert app.shutdown_requested
+        assert not stopped.is_set()
+        response = app.handle(
+            {
+                "command": "attach",
+                "controller_pid": 200,
+                "controller_start": "new",
+                "controller_generation": app.controller_generation,
+            }
+        )
+        assert response["ok"] is False and "exiting" in response["error"]
+        release.set()
+        assert stopped.wait(3)
+        monitor.join(timeout=2)
+        manifest = app.service.manifests.load_current()
+        assert manifest is not None and [n.snapshot for n in manifest.chain] == [first]
+        assert (storage / "raw" / second).is_dir()
+        assert app.results[0]["payload"]["kind"] == "cancelled"
+    finally:
+        release.set()
+        app.shutdown_requested = True
+        monitor.join(timeout=5)
+        app.close()
+
+
+@pytest.mark.parametrize("kind", ["mutation", "export"])
+def test_management_job_cancel_preserves_repository(tmp_path: Path, monkeypatch, kind: str) -> None:
+    from endstone_endkeep.repository.mutation import SnapshotMutator
+    from tests.standalone_fixture import build_fixture
+
+    storage = tmp_path / "backups"
+    build_fixture(storage)
+    app = WorkerApplication(storage, priority="background")
+    app.handle({"command": "configure", "settings": _settings()})
+    service = app.service
+    original = service.manifests.load_current()
+    entered, release = threading.Event(), threading.Event()
+    if kind == "mutation":
+        operation = SnapshotMutator._rollover
+
+        def blocked(*args, **kwargs):
+            entered.set()
+            assert release.wait(5)
+            return operation(*args, **kwargs)
+
+        monkeypatch.setattr(SnapshotMutator, "_rollover", blocked)
+        assert service.start_mutation("rollover", "20261006-163000", original.generation)
+    else:
+        from endstone_endkeep.offline import cli
+
+        operation = cli.command_restore
+
+        def blocked(*args, **kwargs):
+            entered.set()
+            assert release.wait(5)
+            return operation(*args, **kwargs)
+
+        monkeypatch.setattr(cli, "command_restore", blocked)
+        assert service.start_export("20261006-163000")
+    try:
+        assert entered.wait(2)
+        assert service.request_cancel()
+        release.set()
+        deadline = time.monotonic() + 3
+        result = None
+        while result is None and time.monotonic() < deadline:
+            result = service.poll()
+            time.sleep(0.01)
+        assert result is not None and result.kind == "cancelled"
+        assert service.manifests.load_current() == original
+        assert not (storage / "exports" / "20261006-163000").exists()
+    finally:
+        release.set()
+        app.close()
+
+
+@pytest.mark.parametrize(
+    "point,new_committed",
+    [
+        ("after_manifest_fsync", False),
+        ("after_manifest_rename", False),
+        ("after_manifest_dir_fsync", False),
+        ("after_head_fsync", False),
+        ("after_commit", True),
+    ],
+)
+def test_process_kill_during_commit_recovers_and_restores(tmp_path: Path, point: str, new_committed: bool) -> None:
+    import subprocess
+    import sys
+
+    from endstone_endkeep.offline.cli import command_restore
+    from endstone_endkeep.repository.lock import RepositoryLock
+    from endstone_endkeep.repository.manifest import ManifestStore
+    from endstone_endkeep.repository.recovery import StartupRecovery
+    from endstone_endkeep.repository.verify import RepositoryVerifier
+
+    storage, ready = tmp_path / "backups", tmp_path / "ready"
+    process = subprocess.Popen(
+        [sys.executable, "-m", "tests.transaction_crash_fixture", str(storage), point, str(ready)],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        cwd=Path(__file__).resolve().parents[1],
+    )
+    try:
+        deadline = time.monotonic() + 15
+        while not ready.exists() and process.poll() is None and time.monotonic() < deadline:
+            time.sleep(0.02)
+        assert ready.exists(), "child must reach the requested on-disk commit boundary"
+        process.kill()
+        process.communicate(timeout=5)
+        assert process.returncode != 0
+    finally:
+        if process.poll() is None:
+            process.kill()
+            process.communicate(timeout=5)
+    manifests = ManifestStore(storage / "repo")
+    objects = ObjectStore(storage / "repo", compression_level=6, compression_threads=1)
+    with RepositoryLock(storage / "worker-owner"), RepositoryLock(storage / "repo"):
+        recovery = StartupRecovery(storage, manifests, objects).run()
+    current = manifests.load_current()
+    assert current is not None and current.generation == (3 if new_committed else 2)
+    assert ("20261006-200000" in recovery.pending_raw) is not new_committed
+    RepositoryVerifier(manifests, objects).verify(deep=True)
+    assert command_restore(storage / "repo", tmp_path / "restored", None) == 0
+    assert (tmp_path / "restored" / "db").is_dir()
+
+
+def test_cancel_reaches_merge_that_deletes_every_record() -> None:
+    from endstone_endkeep.logical.format import DeltaOperation
+    from endstone_endkeep.logical.merge import apply_delta
+
+    consumed = 0
+    total = 32768
+
+    def state():
+        nonlocal consumed
+        for index in range(total):
+            consumed += 1
+            yield index.to_bytes(4, "big"), b"value"
+
+    def cancel():
+        if consumed >= 16384:
+            raise JobCancelled("stop during non-emitting merge")
+
+    deletions = (DeltaOperation.delete(index.to_bytes(4, "big")) for index in range(total))
+    with pytest.raises(JobCancelled):
+        list(apply_delta(state(), deletions, cancel_check=cancel))
+    assert consumed < total
+
+
+@pytest.mark.parametrize("stage", ["logicalize", "commit"])
+def test_cancelled_raw_can_retry_in_same_service(tmp_path: Path, monkeypatch, stage: str) -> None:
+    storage = tmp_path / "backups"
+    app = WorkerApplication(storage, priority="background")
+    app.handle({"command": "configure", "settings": _settings()})
+    snapshot = "20261014-120000"
+    add_raw(storage, snapshot, [(b"a", b"value")], b"world")
+    service = app.service
+    entered, release = threading.Event(), threading.Event()
+    if stage == "logicalize":
+        create = service.logicalizer._create_base
+
+        def blocked(*args, **kwargs):
+            entered.set()
+            assert release.wait(5)
+            return create(*args, **kwargs)
+
+        monkeypatch.setattr(service.logicalizer, "_create_base", blocked)
+    else:
+        commit = RepositoryTransaction.commit
+
+        def blocked(transaction, manifest):
+            def pause(point):
+                if point == "after_manifest_fsync":
+                    entered.set()
+                    assert release.wait(5)
+
+            transaction.fault_hook = pause
+            return commit(transaction, manifest)
+
+        monkeypatch.setattr(RepositoryTransaction, "commit", blocked)
+    try:
+        assert service.start_maintenance("LOGIC_ONLY")
+        assert entered.wait(2)
+        assert service.request_cancel()
+        release.set()
+        for kind in ("cancelled", "maintenance"):
+            deadline = time.monotonic() + 3
+            result = None
+            while result is None and time.monotonic() < deadline:
+                result = service.poll()
+                time.sleep(0.01)
+            assert result is not None and result.kind == kind
+            if kind == "cancelled":
+                assert service.start_maintenance("LOGIC_ONLY")
+            else:
+                assert not result.maintenance.failures
+                assert [r.snapshot_id for r in result.maintenance.committed] == (
+                    [snapshot] if stage == "logicalize" else []
+                )
+                assert [n.snapshot for n in service.manifests.load_current().chain] == [snapshot]
+                assert not (storage / "raw" / snapshot).exists()
+    finally:
+        release.set()
+        app.close()
diff --git a/tests/test_worker.py b/tests/test_worker.py
index 69d44997..03fade3a 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -1,6 +1,9 @@
 from __future__ import annotations
 
 import os
+import subprocess
+import sys
+import threading
 import time
 from pathlib import Path
 from types import SimpleNamespace
@@ -8,10 +11,19 @@
 
 import pytest
 
-from endstone_endkeep.worker.client import RepositoryWorkerClient, WorkerError
+from endstone_endkeep.worker.client import RepositoryWorkerClient, WorkerError, WorkerTimeout
+from endstone_endkeep.worker.process import process_identity
+from endstone_endkeep.worker.protocol import RuntimeInfo
 from endstone_endkeep.worker.server import WorkerApplication
 
 
+def _poll_with_retry(client: RepositoryWorkerClient):
+    try:
+        return client.poll()
+    except WorkerTimeout:
+        return None
+
+
 def _settings(*, verify_mode: str = "normal") -> dict[str, Any]:
     return {
         "compression_level": 6,
@@ -44,7 +56,7 @@ def test_worker_starts_reconnects_and_runs_normal_verify(tmp_path: Path, monkeyp
         deadline = time.monotonic() + 5.0
         while time.monotonic() < deadline and maintenance_result is None:
             time.sleep(0.05)
-            maintenance_result = second.poll()
+            maintenance_result = _poll_with_retry(second)
 
         assert maintenance_result is not None
         maintenance_result_id, maintenance_payload = maintenance_result
@@ -58,7 +70,7 @@ def test_worker_starts_reconnects_and_runs_normal_verify(tmp_path: Path, monkeyp
         deadline = time.monotonic() + 5.0
         while time.monotonic() < deadline and result is None:
             time.sleep(0.05)
-            result = second.poll()
+            result = _poll_with_retry(second)
 
         assert result is not None
         result_id, payload = result
@@ -67,7 +79,7 @@ def test_worker_starts_reconnects_and_runs_normal_verify(tmp_path: Path, monkeyp
 
         # Results are non-destructive until the controller confirms that it
         # handled them, so a short poll timeout cannot lose completion state.
-        repeated = second.poll()
+        repeated = _poll_with_retry(second)
         assert repeated is not None
         assert repeated[0] == result_id
         assert repeated[1] == payload
@@ -89,7 +101,7 @@ def test_worker_starts_reconnects_and_runs_normal_verify(tmp_path: Path, monkeyp
             adopted_new_plugin.shutdown()
 
         assert adopted_new_plugin.ack_result(result_id) is True
-        assert third.poll() is None
+        assert _poll_with_retry(third) is None
         assert second.refresh()["job"]["state"] == "idle"
 
         if os.name == "posix":
@@ -123,7 +135,7 @@ def test_full_maintenance_honors_configured_deep_verify(tmp_path: Path) -> None:
         deadline = time.monotonic() + 5.0
         while time.monotonic() < deadline and result is None:
             time.sleep(0.05)
-            result = client.poll()
+            result = _poll_with_retry(client)
 
         assert result is not None
         result_id, payload = result
@@ -212,3 +224,410 @@ def fake_rpc(payload, *, timeout=None):
     assert client.POLL_TIMEOUT_SECONDS <= 0.01
     assert client.ACK_TIMEOUT_SECONDS <= 0.01
     assert client.AUTO_CONTROL_TIMEOUT_SECONDS <= 0.02
+
+
+def test_detach_preserves_worker_and_reconnects(tmp_path: Path) -> None:
+    storage = tmp_path / "backups"
+    first = RepositoryWorkerClient.connect_or_start(storage, priority="background")
+    try:
+        first.configure(_settings())
+        first.detach()
+        second = RepositoryWorkerClient.connect_or_start(storage, priority="background")
+        assert second.runtime.instance_id == first.runtime.instance_id
+        assert second.runtime.pid == first.runtime.pid
+        assert second.start_verify(deep=False)
+        deadline = time.monotonic() + 5
+        result = None
+        while time.monotonic() < deadline and result is None:
+            result = _poll_with_retry(second)
+            time.sleep(0.05)
+        assert result is not None
+        assert second.ack_result(result[0])
+        second.shutdown()
+        RepositoryWorkerClient._wait_for_exit(
+            first.runtime.pid, first.runtime_path, process_start=first.runtime.process_start
+        )
+    finally:
+        if RepositoryWorkerClient._runtime_alive(first.runtime):
+            try:
+                first.shutdown()
+            except WorkerError:
+                pass
+
+
+def test_orphan_worker_exits_with_persisted_unacked_result(tmp_path: Path, monkeypatch) -> None:
+    storage = tmp_path / "backups"
+    app = WorkerApplication(storage, priority="background", controller_pid=1234, controller_start="old")
+    app._queue_result({"kind": "maintenance", "request_id": "saved"})
+    stopped = threading.Event()
+    app.set_stop_callback(stopped.set)
+    monkeypatch.setattr("endstone_endkeep.worker.server.process_identity", lambda pid: None)
+    thread = threading.Thread(target=app.monitor, daemon=True)
+    thread.start()
+    assert stopped.wait(3), "orphan must exit even with an unacknowledged result"
+    thread.join(timeout=2)
+    next_app = WorkerApplication(storage, priority="background")
+    assert list(next_app.results)[0]["payload"]["request_id"] == "saved"
+    assert next_app.handle({"command": "ack_result", "result_id": 1})["accepted"]
+    next_app.close()
+    assert not WorkerApplication(storage, priority="background").results
+
+
+def test_worker_owner_attach_rejects_live_foreign_controller(tmp_path: Path, monkeypatch) -> None:
+    app = WorkerApplication(tmp_path / "backups", priority="background", controller_pid=100, controller_start="first")
+    monkeypatch.setattr(
+        "endstone_endkeep.worker.server.process_identity",
+        lambda pid: {100: "first", 200: "second"}.get(pid),
+    )
+    response = app.handle(
+        {
+            "command": "attach",
+            "controller_pid": 200,
+            "controller_start": "second",
+            "controller_generation": app.controller_generation,
+        }
+    )
+    assert response["ok"] is False
+    assert app.controller_pid == 100
+    monkeypatch.setattr(
+        "endstone_endkeep.worker.server.process_identity",
+        lambda pid: {200: "second"}.get(pid),
+    )
+    response = app.handle(
+        {
+            "command": "attach",
+            "controller_pid": 200,
+            "controller_start": "second",
+            "controller_generation": app.controller_generation,
+        }
+    )
+    assert response["ok"] is False and "exiting" in response["error"]
+    assert app.controller_pid == 100
+    assert app.shutdown_requested
+
+
+def test_worker_reconnect_recovers_from_transient_rpc_timeout(tmp_path: Path, monkeypatch) -> None:
+    storage = tmp_path / "backups"
+    first = RepositoryWorkerClient.connect_or_start(storage, priority="background")
+    original_rpc = RepositoryWorkerClient._rpc
+    count = 0
+
+    def flaky_rpc(self, payload, *, timeout=None):
+        nonlocal count
+        if payload.get("command") == "ping" and count == 0:
+            count += 1
+            raise WorkerTimeout("temporary unresponsive worker")
+        return original_rpc(self, payload, timeout=timeout)
+
+    try:
+        monkeypatch.setattr(RepositoryWorkerClient, "_rpc", flaky_rpc)
+        second = RepositoryWorkerClient.connect_or_start(storage, priority="background")
+        assert count == 1
+        assert second.runtime.instance_id == first.runtime.instance_id
+        second.shutdown()
+    finally:
+        monkeypatch.setattr(RepositoryWorkerClient, "_rpc", original_rpc)
+        if RepositoryWorkerClient._runtime_alive(first.runtime):
+            first.shutdown()
+
+
+def test_worker_unresponsive_does_not_start_second_writer(tmp_path: Path, monkeypatch) -> None:
+    storage = tmp_path / "backups"
+    first = RepositoryWorkerClient.connect_or_start(storage, priority="background")
+    monkeypatch.setattr(
+        RepositoryWorkerClient, "_rpc", lambda *a, **kw: (_ for _ in ()).throw(WorkerTimeout("blocked"))
+    )
+    monkeypatch.setattr("endstone_endkeep.worker.client.time.sleep", lambda unused: None)
+    with pytest.raises(WorkerError, match="four RPC attempts"):
+        RepositoryWorkerClient.connect_or_start(storage, priority="background")
+    assert RepositoryWorkerClient._load_runtime(first.runtime_path).instance_id == first.runtime.instance_id
+    monkeypatch.undo()
+    first.shutdown()
+
+
+def test_reused_pid_does_not_count_as_same_runtime(tmp_path: Path, monkeypatch) -> None:
+    runtime = RuntimeInfo(
+        protocol=1,
+        version="x",
+        pid=12345,
+        port=123,
+        token="x",
+        instance_id="old",
+        storage_root=str(tmp_path),
+        priority="background",
+        process_start="original",
+    )
+    monkeypatch.setattr("endstone_endkeep.worker.client.process_identity", lambda pid: "reused")
+    assert not RepositoryWorkerClient._runtime_alive(runtime)
+
+
+def test_worker_process_lifetime_lock_prevents_second_writer(tmp_path: Path) -> None:
+    storage = tmp_path / "backups"
+    first = RepositoryWorkerClient.connect_or_start(storage, priority="background")
+    args = [
+        sys.executable,
+        "-m",
+        "endstone_endkeep.worker",
+        "--storage-root",
+        str(storage),
+        "--runtime-file",
+        str(storage / "worker-runtime.json"),
+        "--priority",
+        "background",
+        "--controller-pid",
+        str(os.getpid()),
+        "--controller-start",
+        process_identity(os.getpid()) or "",
+        "--launch-id",
+        "competing-worker-test",
+    ]
+    try:
+        second = subprocess.run(args, capture_output=True, text=True, timeout=10)
+        assert second.returncode != 0, second.stdout
+        assert RepositoryWorkerClient._load_runtime(first.runtime_path).instance_id == first.runtime.instance_id
+        assert RepositoryWorkerClient._runtime_alive(first.runtime)
+    finally:
+        first.shutdown()
+
+
+def test_attach_cannot_revive_worker_after_exit_has_started(tmp_path: Path, monkeypatch) -> None:
+    app = WorkerApplication(tmp_path / "backups", priority="background", controller_pid=100, controller_start="dead")
+    monkeypatch.setattr("endstone_endkeep.worker.server.process_identity", lambda pid: "new" if pid == 200 else None)
+    stopping = threading.Event()
+    release = threading.Event()
+
+    def stop():
+        stopping.set()
+        release.wait(3)
+
+    app.set_stop_callback(stop)
+    thread = threading.Thread(target=app.monitor, daemon=True)
+    thread.start()
+    try:
+        assert stopping.wait(2)
+        response = app.handle(
+            {
+                "command": "attach",
+                "controller_pid": 200,
+                "controller_start": "new",
+                "controller_generation": app.controller_generation,
+            }
+        )
+        assert response["ok"] is False, "a worker committed to exit cannot promise ownership to a new controller"
+        assert "exiting" in response["error"]
+        assert app.handle({"command": "shutdown"})["ok"] is True
+    finally:
+        release.set()
+        thread.join(timeout=3)
+        app.close()
+
+
+@pytest.mark.parametrize("abandonment", ["controller-death", "unload", "reload"])
+def test_running_job_cancels_after_abandonment_but_survives_reload(
+    tmp_path: Path, monkeypatch, abandonment: str
+) -> None:
+    from tests.standalone_fixture import add_raw
+
+    storage = tmp_path / "backups"
+    app = WorkerApplication(storage, priority="background", controller_pid=100, controller_start="owner")
+    monkeypatch.setattr("endstone_endkeep.worker.server.process_identity", lambda pid: "owner")
+    app.handle({"command": "configure", "settings": _settings()})
+    add_raw(storage, "20261014-120000", [(b"a", b"one"), (b"b", b"two")], b"world")
+    entered = threading.Event()
+    release = threading.Event()
+    stopped = threading.Event()
+    service = app.service
+    assert service is not None
+    logicalize = service.logicalizer.logicalize
+
+    def blocked_logicalize(*args, **kwargs):
+        entered.set()
+        assert release.wait(5)
+        return logicalize(*args, **kwargs)
+
+    monkeypatch.setattr(service.logicalizer, "logicalize", blocked_logicalize)
+    assert app.handle({"command": "start_maintenance", "mode": "LOGIC_ONLY"})["accepted"]
+    assert entered.wait(2)
+    if abandonment == "controller-death":
+        monkeypatch.setattr("endstone_endkeep.worker.server.process_identity", lambda pid: None)
+    else:
+        monkeypatch.setattr(
+            "endstone_endkeep.worker.server._HANDOFF_GRACE_SECONDS", 0 if abandonment == "unload" else 15
+        )
+        app.handle({"command": "detach", "controller_generation": app.controller_generation})
+        if abandonment == "reload":
+            assert app.handle(
+                {
+                    "command": "attach",
+                    "controller_pid": 100,
+                    "controller_start": "owner",
+                    "controller_generation": app.controller_generation,
+                }
+            )["ok"]
+            assert app.service is service
+    app.set_stop_callback(stopped.set)
+    monitor = threading.Thread(target=app.monitor, daemon=True)
+    monitor.start()
+    try:
+        if abandonment != "reload":
+            deadline = time.monotonic() + 2
+            while not app.shutdown_requested and time.monotonic() < deadline:
+                time.sleep(0.02)
+            assert app.shutdown_requested
+        assert not stopped.is_set(), "shutdown must wait until the job reaches a safe checkpoint"
+        assert not app.handle({"command": "start_verify"}).get("accepted", False)
+        release.set()
+        if abandonment == "reload":
+            deadline = time.monotonic() + 4
+            while not app.results and time.monotonic() < deadline:
+                time.sleep(0.02)
+            assert app.results
+            assert not stopped.is_set()
+            app.shutdown_requested = True
+        assert stopped.wait(4)
+        monitor.join(timeout=2)
+        manifest = service.manifests.load_current()
+        if abandonment == "reload":
+            assert manifest is not None
+            assert [node.snapshot for node in manifest.chain] == ["20261014-120000"]
+        else:
+            assert manifest is None
+            assert (storage / "raw" / "20261014-120000").is_dir()
+        app.close()
+        restored_app = WorkerApplication(storage, priority="background")
+        payload = restored_app.results[0]["payload"]
+        assert payload["kind"] == ("maintenance" if abandonment == "reload" else "cancelled")
+        if abandonment == "reload":
+            assert payload["maintenance"]["failures"] == []
+            assert payload["maintenance"]["committed"][0]["snapshot_id"] == "20261014-120000"
+        restored_app.close()
+    finally:
+        release.set()
+        app.shutdown_requested = True
+        monitor.join(timeout=5)
+        app.close()
+
+
+def test_monitor_flush_failure_still_triggers_safe_exit(tmp_path: Path, monkeypatch) -> None:
+    app = WorkerApplication(tmp_path / "backups", priority="background")
+    app._queue_result({"kind": "maintenance", "request_id": "durable"})
+    assert app.handle({"command": "ack_result", "result_id": 1})["accepted"]
+    stopped = threading.Event()
+    app.set_stop_callback(stopped.set)
+
+    def fail_flush(*args):
+        raise OSError("injected worker result flush failure")
+
+    monkeypatch.setattr(app, "_save_results", fail_flush)
+    monitor = threading.Thread(target=app.monitor, daemon=True)
+    monitor.start()
+    try:
+        assert stopped.wait(2), "a failed monitor must shut down the worker instead of leaving an orphan"
+        assert not app.handle({"command": "configure", "settings": _settings()})["ok"]
+    finally:
+        monitor.join(timeout=2)
+        monkeypatch.undo()
+        app.close()
+
+
+def test_close_drains_service_even_if_result_flush_fails(tmp_path: Path, monkeypatch) -> None:
+    app = WorkerApplication(tmp_path / "backups", priority="background")
+    closed = []
+    app.service = SimpleNamespace(close=lambda: closed.append(True), poll=lambda: None)
+    app._results_dirty = True
+    monkeypatch.setattr(app, "_save_results", lambda *args: (_ for _ in ()).throw(OSError("flush failed")))
+    with pytest.raises(OSError, match="flush failed"):
+        app.close()
+    assert closed == [True], "failed result persistence must not skip repository task drain"
+
+
+def test_delayed_old_detach_cannot_expire_new_controller(tmp_path: Path, monkeypatch) -> None:
+    app = WorkerApplication(tmp_path / "backups", priority="background", controller_pid=100, controller_start="owner")
+    monkeypatch.setattr("endstone_endkeep.worker.server.process_identity", lambda pid: "owner")
+    first = app.handle(
+        {
+            "command": "attach",
+            "controller_pid": 100,
+            "controller_start": "owner",
+            "controller_generation": app.controller_generation,
+        }
+    )
+    previous = first["status"]["worker"].get("controller_generation", 0)
+    app.handle(
+        {
+            "command": "attach",
+            "controller_pid": 100,
+            "controller_start": "owner",
+            "controller_generation": app.controller_generation,
+        }
+    )
+    response = app.handle({"command": "detach", "controller_generation": previous})
+    assert not response["ok"], "a delayed disable request must not detach the replacement plugin"
+    assert app.detached_until is None
+
+
+def test_reconnect_retries_transient_attach_timeout(tmp_path: Path, monkeypatch) -> None:
+    runtime = RuntimeInfo(1, "test", 100, 123, "token", "instance", str(tmp_path), "background", "worker")
+    monkeypatch.setattr(RepositoryWorkerClient, "_load_runtime", lambda path: runtime)
+    monkeypatch.setattr(
+        "endstone_endkeep.worker.client.process_identity", lambda pid: "worker" if pid == 100 else "owner"
+    )
+    monkeypatch.setattr("endstone_endkeep.worker.client.package_version", lambda: "test")
+    status = {"worker": {"pid": 100, "controller_start": "owner", "controller_generation": 1}}
+    missed = False
+
+    def rpc(self, payload, *, timeout=None):
+        nonlocal missed
+        if payload["command"] == "attach" and not missed:
+            missed = True
+            raise WorkerTimeout("temporary attach timeout")
+        return {"ok": True, "status": status}
+
+    monkeypatch.setattr(RepositoryWorkerClient, "_rpc", rpc)
+    client = RepositoryWorkerClient.connect_or_start(tmp_path, priority="background")
+    assert missed
+    assert client.runtime.instance_id == runtime.instance_id
+
+
+def test_failed_result_write_is_preserved_for_final_flush(tmp_path: Path, monkeypatch) -> None:
+    storage = tmp_path / "backups"
+    app = WorkerApplication(storage, priority="background")
+    original_save = app._save_results
+    first = True
+
+    def fail_once(results, acked):
+        nonlocal first
+        if first:
+            first = False
+            raise OSError("temporary result write failure")
+        original_save(results, acked)
+
+    monkeypatch.setattr(app, "_save_results", fail_once)
+    with pytest.raises(OSError, match="temporary result write failure"):
+        app._queue_result({"kind": "maintenance", "request_id": "completed"})
+    app.close()
+    restored = WorkerApplication(storage, priority="background")
+    assert restored.results[0]["payload"]["request_id"] == "completed"
+    restored.close()
+
+
+def test_worker_runtime_is_removed_even_when_final_flush_fails(tmp_path: Path, monkeypatch) -> None:
+    from endstone_endkeep.worker import server
+
+    path = tmp_path / "worker-runtime.json"
+    monkeypatch.setattr(server._ThreadingServer, "serve_forever", lambda *args, **kwargs: None)
+    monkeypatch.setattr(server.threading.Thread, "start", lambda self: None)
+    monkeypatch.setattr(server.WorkerApplication, "close", lambda self: (_ for _ in ()).throw(OSError("flush failed")))
+    with pytest.raises(OSError, match="flush failed"):
+        server._run_worker(tmp_path, path, "background", 100, "owner", "test-launch")
+    assert not path.exists(), "failed final flush must not leave runtime pointing at an exited worker"
+
+
+def test_delayed_attach_retry_cannot_replace_successful_attachment(tmp_path: Path, monkeypatch) -> None:
+    app = WorkerApplication(tmp_path / "backups", priority="background", controller_pid=100, controller_start="owner")
+    monkeypatch.setattr("endstone_endkeep.worker.server.process_identity", lambda pid: "owner")
+    request = {"command": "attach", "controller_pid": 100, "controller_start": "owner", "controller_generation": 0}
+    response = app.handle(request)
+    acquired = response["status"]["worker"]["controller_generation"]
+    assert not app.handle(request)["ok"], "the delayed earlier attempt must not advance ownership again"
+    assert app.handle({"command": "detach", "controller_generation": acquired})["ok"]
diff --git a/tests/transaction_crash_fixture.py b/tests/transaction_crash_fixture.py
new file mode 100644
index 00000000..a5095c3a
--- /dev/null
+++ b/tests/transaction_crash_fixture.py
@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+from endstone_endkeep.repository.lock import RepositoryLock
+from endstone_endkeep.repository.logicalize import Logicalizer
+from endstone_endkeep.repository.transaction import RepositoryTransaction
+from tests.standalone_fixture import add_raw, build_fixture
+
+
+def main() -> None:
+    storage, point, ready = Path(sys.argv[1]), sys.argv[2], Path(sys.argv[3])
+    build_fixture(storage)
+    add_raw(storage, "20261006-200000", [(b"a", b"last"), (b"c", b"3")], b"last")
+    original = RepositoryTransaction.commit
+
+    def commit(transaction, manifest):
+        def pause(current):
+            if current == point:
+                ready.write_text(current, encoding="ascii")
+                # Parent kills this process; no Python cleanup can run.
+                import threading
+
+                threading.Event().wait(30)
+                raise TimeoutError("parent did not terminate crash fixture")
+
+        transaction.fault_hook = pause
+        return original(transaction, manifest)
+
+    RepositoryTransaction.commit = commit
+    with RepositoryLock(storage / "worker-owner"), RepositoryLock(storage / "repo"):
+        Logicalizer(storage, compression_level=6, compression_threads=1).logicalize(storage / "raw" / "20261006-200000")
+
+
+if __name__ == "__main__":
+    main()