feat: complete transfer and eviction execution

e2e/README.md

@@ -27,11 +27,12 @@ deadline. `down.sh` removes only the five exact Compose projects and the
 labelled E2E network; it intentionally retains all bind-mounted runtime state.
 
 Set `E2E_RUN_TRANSFER=1` to follow route verification with a real
-cache-1 → archive-1 transfer followed by archive-1 → cache-2 unarchive. The
+cache-1 → archive-1 transfer, covered cache-1 eviction, and
-scenario creates a deterministic one-file torrent in the isolated cache
+archive-1 → cache-2 unarchive. The scenario creates a deterministic one-file
-qBittorrent, submits both jobs through the test HTTP adapter, waits for all five
+torrent in the isolated cache qBittorrent, uses fresh preview revisions for all
-durable steps in each direction, and verifies target qBittorrent selections,
+jobs, waits for the durable five/three/five-step flows, and verifies target
-retained source entries, and target file digests.
+qBittorrent selections, archive retention, safe cache removal, and target file
+digests.
 
 The Syncthing 2.1.2 and LinuxServer qBittorrent multi-platform image indexes are
 digest-pinned. Runtime secrets are generated with mode 0600 and ignored by

e2e/scenarios/archive_happy.py

@@ -9,8 +9,6 @@ import sys
 import time
 import urllib.parse
 import urllib.request
-import uuid
-from datetime import datetime, timezone
 
 
 CONTROL = "http://127.0.0.1:18081/test/v1"
@@ -103,30 +101,18 @@ def main() -> int:
         60,
         f"{args.cache_client}/{args.archive_client} route",
     )
-    job_id = str(uuid.uuid4())
+    preview = post_json(f"{CONTROL}/jobs/preview", {
-    now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+        "operation": args.operation,
-    definition = {
+        "source_client_id": args.source_client,
-        "jobId": job_id,
+        "target_client_id": args.target_client,
-        "idempotencyKey": str(uuid.uuid4()),
+        "resource_id": {"info_hash_v1_hex": INFO_HASH},
-        "operation": (
+    })
-            "JOB_OPERATION_ARCHIVE"
+    definition = preview["definition"]
-            if args.operation == "archive"
+    job_id = definition["job_id"]
-            else "JOB_OPERATION_UNARCHIVE"
+    created = post_json(f"{CONTROL}/jobs", {
-        ),
+        "preview_revision": preview["preview_revision"],
-        "resourceId": {"infoHashV1Hex": INFO_HASH},
+        "definition": definition,
-        "resourceDisplayName": "fixture.bin",
+    })
-        "createdAt": now,
-        "transfer": {
-            "sourceClientId": args.source_client,
-            "targetClientId": args.target_client,
-            "routeId": route["route_id"],
-            "requestedFiles": {"ranges": [{"first": 0, "last": 0}]},
-            "transferDeltaFiles": {"ranges": [{"first": 0, "last": 0}]},
-            "requestedLogicalBytes": str(LOGICAL_BYTES),
-            "transferDeltaLogicalBytes": str(LOGICAL_BYTES),
-        },
-    }
-    created = post_json(f"{CONTROL}/jobs", definition)
     if created["job_id"] != job_id:
         raise RuntimeError("control returned the wrong job")
 

e2e/scenarios/evict_happy.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Drive a covered cache eviction through the bot-free HTTP adapter."""
+
+from __future__ import annotations
+
+import json
+import sys
+import time
+import urllib.request
+
+
+CONTROL = "http://127.0.0.1:18081/test/v1"
+INFO_HASH = "330be0cb7c2201135a2de63b28e77993745ff688"
+
+
+def get_json(url: str):
+    with urllib.request.urlopen(url, timeout=35) as response:
+        return json.load(response)
+
+
+def post_json(url: str, value: object):
+    request = urllib.request.Request(
+        url,
+        data=json.dumps(value, separators=(",", ":")).encode(),
+        method="POST",
+        headers={"Content-Type": "application/json"},
+    )
+    with urllib.request.urlopen(request, timeout=35) as response:
+        return json.load(response)
+
+
+def main() -> int:
+    preview = post_json(f"{CONTROL}/jobs/preview", {
+        "operation": "evict_cache",
+        "cache_client_id": "cache-1",
+        "resource_id": {"info_hash_v1_hex": INFO_HASH},
+    })
+    definition = preview["definition"]
+    job_id = definition["job_id"]
+    post_json(f"{CONTROL}/jobs", {
+        "preview_revision": preview["preview_revision"],
+        "definition": definition,
+    })
+    deadline = time.monotonic() + 180
+    while time.monotonic() < deadline:
+        job = get_json(f"{CONTROL}/jobs/{job_id}")
+        if job["state"] == "JOB_STATE_SUCCEEDED" and job["committed"]:
+            print(json.dumps({
+                "job_id": job_id,
+                "state": job["state"],
+                "committed": job["committed"],
+            }, sort_keys=True))
+            return 0
+        if job["state"] in {"JOB_STATE_FAILED", "JOB_STATE_CANCELLED"}:
+            raise RuntimeError(f"eviction did not succeed: {job}")
+        time.sleep(0.5)
+    raise RuntimeError("timed out waiting for cache eviction")
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except Exception as exc:
+        print(f"eviction happy-path failed: {exc}", file=sys.stderr)
+        raise

e2e/scripts/archive-happy.sh

@@ -63,6 +63,16 @@ if [[ "$target_digest" != \
 fi
 printf 'archive filesystem content verified: sha256=%s\n' "$target_digest"
 
+compose_control exec -T control \
+    python /e2e/scenarios/evict_happy.py
+wait_qb_absent cache-1
+if compose_node cache-1 exec -T qbittorrent \
+    test -e /downloads/fixture.bin; then
+    printf 'eviction left the unshared cache file behind\n' >&2
+    exit 1
+fi
+printf 'covered cache eviction verified\n'
+
 compose_node cache-2 exec -T qbittorrent curl -fsS \
     -X POST http://127.0.0.1:8080/api/v2/torrents/delete \
     --data-urlencode "hashes=$info_hash" \

examples/client.toml

@@ -13,6 +13,16 @@ reconnect_max = "60s"
 reconnect_reset_after = "60s"
 reconnect_jitter = true
 
+[jobs]
+# Mark a job stalled after it has made no progress for this long.
+stall_after = "30m"
+# Upper bound for qBittorrent's stopped full recheck.
+verification_timeout = "30m"
+# Poll qBittorrent and Syncthing at this interval while a step is active.
+poll_interval = "1s"
+# Space that must remain free after a worst-case copy fallback.
+free_space_reserve_bytes = 1073741824
+
 [backup]
 interval = "6h"
 recent = 12

src/archive_clients/config.py

@@ -80,6 +80,9 @@ class ConnectionConfig:
 @dataclass(frozen=True)
 class JobsConfig:
     stall_after: float = 30 * 60
+    verification_timeout: float = 30 * 60
+    poll_interval: float = 1
+    free_space_reserve_bytes: int = 1024 * 1024 * 1024
 
 
 @dataclass(frozen=True)
@@ -217,8 +220,27 @@ def _connection(value: Any) -> ConnectionConfig:
 def _jobs(value: Any) -> JobsConfig:
     if not isinstance(value, dict):
         raise ConfigError("jobs must be a table")
-    _keys(value, {"stall_after"}, "jobs")
+    _keys(
-    return JobsConfig(stall_after=_duration(value.get("stall_after", "30m")))
+        value,
+        {
+            "stall_after",
+            "verification_timeout",
+            "poll_interval",
+            "free_space_reserve_bytes",
+        },
+        "jobs",
+    )
+    return JobsConfig(
+        stall_after=_duration(value.get("stall_after", "30m")),
+        verification_timeout=_duration(
+            value.get("verification_timeout", "30m")
+        ),
+        poll_interval=_duration(value.get("poll_interval", "1s")),
+        free_space_reserve_bytes=_positive_int(
+            value.get("free_space_reserve_bytes", 1024 * 1024 * 1024),
+            "jobs.free_space_reserve_bytes",
+        ),
+    )
 
 
 def _backup(value: Any) -> BackupConfig:

src/archive_clients/daemon.py

@@ -105,6 +105,11 @@ class ArchiveClientDaemon:
                 route_path=self._route_path,
                 syncthing_transport=self.routes.transport,
                 sparse_supported=all(probe.sparse_files for probe in probes),
+                poll_interval=config.jobs.poll_interval,
+                verification_timeout=config.jobs.verification_timeout,
+                free_space_reserve_bytes=(
+                    config.jobs.free_space_reserve_bytes
+                ),
             )
             if resource_reader is not None and self.routes is not None
             else None
@@ -415,9 +420,13 @@ class ArchiveClientDaemon:
         elif (
             accepted is not None
             and accepted_for_execution
-            and command.WhichOneof("payload") in {"assign_job", "execute_step"}
+            and command.WhichOneof("payload") in {
+                "assign_job", "execute_step", "cancel_job"
+            }
             and self.jobs is not None
         ):
+            if command.WhichOneof("payload") == "cancel_job":
+                self.jobs.request_cancel(command.cancel_job.job_id)
             self._schedule_job_command(
                 command,
                 envelope.message_id,
@@ -445,9 +454,13 @@ class ArchiveClientDaemon:
                     command, "", outbound, command_tasks
                 )
             elif (
-                command.WhichOneof("payload") in {"assign_job", "execute_step"}
+                command.WhichOneof("payload") in {
+                    "assign_job", "execute_step", "cancel_job"
+                }
                 and self.jobs is not None
             ):
+                if command.WhichOneof("payload") == "cancel_job":
+                    self.jobs.request_cancel(command.cancel_job.job_id)
                 job_commands.append(command)
         if job_commands:
             task = asyncio.create_task(
@@ -543,17 +556,39 @@ class ArchiveClientDaemon:
     ) -> None:
         assert self.jobs is not None
         payload = command.WhichOneof("payload")
+        streamed = False
         if payload == "assign_job":
             events = await asyncio.to_thread(
                 self.jobs.assign, command.assign_job
             )
         elif payload == "execute_step":
+            loop = asyncio.get_running_loop()
+
+            def emit(event):
+                response = new_envelope()
+                response.correlation_id = correlation_id
+                response.job_event.CopyFrom(event)
+                future = asyncio.run_coroutine_threadsafe(
+                    outbound.put(encode(response)), loop
+                )
+                try:
+                    future.result(timeout=30)
+                except Exception:
+                    # The event is already durable in the client DB and will
+                    # be replayed after reconnect.
+                    pass
+
             events = await asyncio.to_thread(
-                self.jobs.execute, command.execute_step
+                self.jobs.execute, command.execute_step, emit
+            )
+            streamed = True
+        elif payload == "cancel_job":
+            events = await asyncio.to_thread(
+                self.jobs.cancel, command.cancel_job
             )
         else:
             raise JobExecutionError("job command payload is unsupported")
-        for event in events:
+        for event in (() if streamed else events):
             response = new_envelope()
             response.correlation_id = correlation_id
             response.job_event.CopyFrom(event)
@@ -880,17 +915,25 @@ class ArchiveClientDaemon:
                 acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_ACCEPTED
         elif command.WhichOneof("payload") == "assign_job":
             definition = command.assign_job.job
+            specification = definition.WhichOneof("spec")
+            assigned_to_client = (
+                specification == "transfer"
+                and self.config.client_id
+                in {
+                    definition.transfer.source_client_id,
+                    definition.transfer.target_client_id,
+                }
+            ) or (
+                specification == "eviction"
+                and self.config.client_id == definition.eviction.cache_client_id
+            )
             if self.jobs is None:
                 acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED
                 acknowledgement.error.code = common_pb2.ERROR_CODE_UNAVAILABLE
                 acknowledgement.error.message = "job executor is unavailable"
             elif (
                 not definition.job_id
-                or definition.WhichOneof("spec") != "transfer"
+                or not assigned_to_client
-                or self.config.client_id not in {
-                    definition.transfer.source_client_id,
-                    definition.transfer.target_client_id,
-                }
             ):
                 acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED
                 acknowledgement.error.code = common_pb2.ERROR_CODE_INVALID_ARGUMENT
@@ -911,6 +954,9 @@ class ArchiveClientDaemon:
                     job_pb2.JOB_STEP_KIND_TARGET_MATERIALIZE,
                     job_pb2.JOB_STEP_KIND_QB_VERIFY,
                     job_pb2.JOB_STEP_KIND_STAGING_CLEANUP,
+                    job_pb2.JOB_STEP_KIND_VERIFY_ARCHIVE_COVERAGE,
+                    job_pb2.JOB_STEP_KIND_QB_REMOVE_ENTRY,
+                    job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK,
                 }
             ):
                 acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED

src/archive_clients/eviction.py

@@ -0,0 +1,242 @@
+"""Idempotent cache eviction without recursive or qB data deletion."""
+
+from __future__ import annotations
+
+import errno
+import hashlib
+import os
+import stat
+from pathlib import Path, PurePosixPath
+from typing import Iterable
+
+from archive_clients.qbittorrent import QBittorrentReader
+from archive_clients.resources import NormalizedResource
+from archive_clients.state import ClientStore
+
+
+class EvictionError(RuntimeError):
+    pass
+
+
+def verify_and_snapshot(
+    *,
+    job_id: str,
+    resource: NormalizedResource,
+    selected_indices: Iterable[int],
+    qb_root: Path,
+    store: ClientStore,
+) -> dict[str, object]:
+    selected = set(selected_indices)
+    by_index = {item.file_index: item for item in resource.files}
+    if not selected or any(index not in by_index for index in selected):
+        raise EvictionError("eviction file selection is invalid")
+    files = []
+    for index in sorted(selected):
+        item = by_index[index]
+        if not item.selected or item.completed_bytes != item.logical_bytes:
+            raise EvictionError(
+                f"cache file {index} is not selected and complete"
+            )
+        relative = _relative(item.canonical_path)
+        path = qb_root.joinpath(*relative.parts)
+        try:
+            metadata = path.lstat()
+        except FileNotFoundError as exc:
+            raise EvictionError(
+                f"cache file disappeared before eviction: {relative}"
+            ) from exc
+        if not stat.S_ISREG(metadata.st_mode):
+            raise EvictionError(
+                f"cache path is not a regular file: {relative}"
+            )
+        if metadata.st_size != item.logical_bytes:
+            raise EvictionError(
+                f"cache file size changed before eviction: {relative}"
+            )
+        files.append(
+            {
+                "file_index": index,
+                "path": relative.as_posix(),
+                "logical_bytes": item.logical_bytes,
+                "device": metadata.st_dev,
+                "inode": metadata.st_ino,
+                "mtime_ns": metadata.st_mtime_ns,
+                "ctime_ns": metadata.st_ctime_ns,
+                "sha256": _sha256(path),
+            }
+        )
+    snapshot = {"files": files}
+    return store.put_job_artifact(job_id, "eviction-snapshot", snapshot)["value"]
+
+
+def remove_qb_entry(
+    *,
+    job_id: str,
+    torrent_hash: str,
+    qbittorrent: QBittorrentReader,
+    store: ClientStore,
+) -> None:
+    marker = store.get_job_artifact(job_id, "qb-entry-removed")
+    if marker is not None:
+        return
+    if qbittorrent.get_resource(torrent_hash) is not None:
+        qbittorrent.delete_entry(torrent_hash)
+    store.put_job_artifact(
+        job_id, "qb-entry-removed", {"torrent_hash": torrent_hash}
+    )
+
+
+def safe_unlink(
+    *,
+    job_id: str,
+    qb_root: Path,
+    qbittorrent: QBittorrentReader,
+    store: ClientStore,
+) -> dict[str, object]:
+    completed = store.get_job_artifact(job_id, "eviction-unlinked")
+    if completed is not None:
+        return completed["value"]
+    snapshot_row = store.get_job_artifact(job_id, "eviction-snapshot")
+    if snapshot_row is None:
+        raise EvictionError("eviction snapshot is missing")
+    snapshot = snapshot_row["value"]
+    files = snapshot.get("files")
+    if not isinstance(files, list):
+        raise EvictionError("eviction snapshot is invalid")
+
+    referenced = _remaining_paths(qbittorrent.list_resources())
+    removed: list[str] = []
+    retained: list[dict[str, str]] = []
+    directories: set[Path] = set()
+    for record in files:
+        if not isinstance(record, dict) or not isinstance(record.get("path"), str):
+            raise EvictionError("eviction file record is invalid")
+        relative = _relative(record["path"])
+        if relative.as_posix() in referenced:
+            retained.append({"path": relative.as_posix(), "reason": "shared"})
+            continue
+        path = qb_root.joinpath(*relative.parts)
+        try:
+            metadata = path.lstat()
+        except FileNotFoundError:
+            removed.append(relative.as_posix())
+            continue
+        if not stat.S_ISREG(metadata.st_mode):
+            retained.append(
+                {"path": relative.as_posix(), "reason": "type-changed"}
+            )
+            continue
+        if (
+            metadata.st_dev != record.get("device")
+            or metadata.st_ino != record.get("inode")
+            or metadata.st_size != record.get("logical_bytes")
+            or metadata.st_mtime_ns != record.get("mtime_ns")
+            or metadata.st_ctime_ns != record.get("ctime_ns")
+            or _sha256(path) != record.get("sha256")
+        ):
+            retained.append(
+                {"path": relative.as_posix(), "reason": "identity-changed"}
+            )
+            continue
+        path.unlink()
+        removed.append(relative.as_posix())
+        parent = path.parent
+        while parent != qb_root:
+            directories.add(parent)
+            parent = parent.parent
+
+    removed_directories: list[str] = []
+    for directory in sorted(
+        directories, key=lambda item: len(item.parts), reverse=True
+    ):
+        try:
+            directory.rmdir()
+            removed_directories.append(
+                directory.relative_to(qb_root).as_posix()
+            )
+        except OSError as exc:
+            if exc.errno not in {errno.ENOTEMPTY, errno.ENOENT}:
+                raise
+    result = {
+        "removed_files": removed,
+        "retained_files": retained,
+        "removed_directories": removed_directories,
+    }
+    return store.put_job_artifact(job_id, "eviction-unlinked", result)["value"]
+
+
+def compensate_materialized_files(
+    *, job_id: str, qb_root: Path, store: ClientStore
+) -> list[str]:
+    """Remove only target files this job created and whose inode still matches."""
+
+    removed: list[str] = []
+    directories: set[Path] = set()
+    for row in store.file_operation_rows(job_id):
+        if row["state"] != "completed" or row["result_json"] is None:
+            continue
+        import json
+
+        intent = json.loads(str(row["intent_json"]))
+        result = json.loads(str(row["result_json"]))
+        if result.get("destination_preexisted"):
+            continue
+        relative_value = intent.get("destination")
+        if not isinstance(relative_value, str):
+            continue
+        relative = _relative(relative_value)
+        path = qb_root.joinpath(*relative.parts)
+        try:
+            metadata = path.lstat()
+        except FileNotFoundError:
+            continue
+        if (
+            not stat.S_ISREG(metadata.st_mode)
+            or metadata.st_dev != result.get("destination_device")
+            or metadata.st_ino != result.get("destination_inode")
+        ):
+            continue
+        path.unlink()
+        removed.append(relative.as_posix())
+        parent = path.parent
+        while parent != qb_root:
+            directories.add(parent)
+            parent = parent.parent
+    for directory in sorted(
+        directories, key=lambda item: len(item.parts), reverse=True
+    ):
+        try:
+            directory.rmdir()
+        except OSError as exc:
+            if exc.errno not in {errno.ENOTEMPTY, errno.ENOENT}:
+                raise
+    return removed
+
+
+def _remaining_paths(resources: Iterable[NormalizedResource]) -> set[str]:
+    return {
+        item.canonical_path
+        for resource in resources
+        for item in resource.files
+    }
+
+
+def _relative(value: str) -> PurePosixPath:
+    candidate = PurePosixPath(value)
+    if (
+        not value
+        or candidate.is_absolute()
+        or ".." in candidate.parts
+        or "." in candidate.parts
+        or "\\" in value
+    ):
+        raise EvictionError("eviction path is unsafe")
+    return candidate
+
+
+def _sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        while chunk := source.read(1024 * 1024):
+            digest.update(chunk)
+    return digest.hexdigest()

src/archive_clients/jobs.py

@@ -3,12 +3,22 @@
 from __future__ import annotations
 
 import hashlib
+import json
+import shutil
+import threading
 import time
 import uuid
 from pathlib import Path, PurePosixPath
 from typing import Callable
 
 from archive_clients.protocol import decode_message, encode_message
+from archive_clients.eviction import (
+    EvictionError,
+    compensate_materialized_files,
+    remove_qb_entry,
+    safe_unlink,
+    verify_and_snapshot,
+)
 from archive_clients.qbittorrent import (
     QBittorrentDownloadAttempt,
     QBittorrentReader,
@@ -37,6 +47,10 @@ class JobExecutionError(RuntimeError):
     pass
 
 
+class JobCancelled(JobExecutionError):
+    pass
+
+
 class ClientJobExecutor:
     def __init__(
         self,
@@ -50,6 +64,8 @@ class ClientJobExecutor:
         syncthing_transport: object,
         sparse_supported: bool,
         poll_interval: float = 1,
+        verification_timeout: float = 30 * 60,
+        free_space_reserve_bytes: int = 1024 * 1024 * 1024,
     ):
         self.client_id = client_id
         self.qbittorrent = qbittorrent
@@ -60,6 +76,12 @@ class ClientJobExecutor:
         self.syncthing_transport = syncthing_transport
         self.sparse_supported = sparse_supported
         self.poll_interval = poll_interval
+        self.verification_timeout = verification_timeout
+        self.free_space_reserve_bytes = free_space_reserve_bytes
+        self._cancel_events: dict[str, threading.Event] = {}
+
+    def request_cancel(self, job_id: str) -> None:
+        self._cancel_events.setdefault(job_id, threading.Event()).set()
 
     def assign(
         self, command: control_pb2.AssignJobCommand
@@ -83,7 +105,9 @@ class ClientJobExecutor:
         return [event]
 
     def execute(
-        self, command: control_pb2.ExecuteStepCommand
+        self,
+        command: control_pb2.ExecuteStepCommand,
+        event_callback: Callable[[control_pb2.JobEvent], None] | None = None,
     ) -> list[control_pb2.JobEvent]:
         definition = self._definition(command.job_id)
         replay = self._replay(
@@ -97,6 +121,9 @@ class ClientJobExecutor:
             control_pb2.JOB_EVENT_TYPE_CLEANUP_REQUIRED,
             control_pb2.JOB_EVENT_TYPE_CANCELLED,
         }:
+            if event_callback is not None:
+                for event in replay:
+                    event_callback(event)
             return replay
         started = replay[0] if replay else self._event(
             definition,
@@ -107,23 +134,124 @@ class ClientJobExecutor:
             committed=(
                 self._committed(command.job_id)
                 or command.step == job_pb2.JOB_STEP_KIND_STAGING_CLEANUP
+                or (
+                    command.step == job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK
+                    and self.store.get_job_artifact(
+                        command.job_id, "qb-entry-removed"
+                    ) is not None
+                )
             ),
             step=command.step,
             step_state=job_pb2.STEP_STATE_RUNNING,
         )
         if not replay:
             self._record(definition, started)
+        if event_callback is not None:
+            event_callback(started)
+        emitted = [started]
+        cursor = started
+        speed_sample = [time.monotonic(), 0]
+
+        def progress(
+            fraction: float,
+            bytes_complete: int = 0,
+            bytes_total: int = 0,
+            detail: str = "",
+        ) -> None:
+            nonlocal cursor
+            now = time.monotonic()
+            elapsed = now - float(speed_sample[0])
+            if fraction < 1 and elapsed < 1:
+                return
+            speed = (
+                max(0, bytes_complete - int(speed_sample[1])) / elapsed
+                if elapsed > 0 else 0
+            )
+            event = self._event(
+                definition,
+                sequence=cursor.sequence + 1,
+                revision=cursor.job_revision + 1,
+                event_type=control_pb2.JOB_EVENT_TYPE_PROGRESS,
+                state=job_pb2.JOB_STATE_RUNNING,
+                committed=cursor.committed,
+                step=command.step,
+                step_state=job_pb2.STEP_STATE_RUNNING,
+            )
+            event.progress.fraction_complete = max(
+                0.0, min(float(fraction), 1.0)
+            )
+            event.progress.bytes_complete = max(0, bytes_complete)
+            event.progress.bytes_total = max(0, bytes_total)
+            event.progress.approximate_bytes_per_second = speed
+            event.progress.detail = detail
+            completed_steps = max(0, _step_number(command.step) - 1)
+            event.progress.overall_fraction_complete = (
+                completed_steps + event.progress.fraction_complete
+            ) / event.progress.display_step_total
+            self._record(definition, event)
+            emitted.append(event)
+            cursor = event
+            speed_sample[:] = [now, bytes_complete]
+            if event_callback is not None:
+                event_callback(event)
         try:
-            result = self._execute_step(definition, command.step)
+            result = self._execute_step(definition, command.step, progress)
+        except JobCancelled as error:
+            cancelling = self._event(
+                definition,
+                sequence=cursor.sequence + 1,
+                revision=cursor.job_revision + 1,
+                event_type=control_pb2.JOB_EVENT_TYPE_CANCELLING,
+                state=job_pb2.JOB_STATE_CANCELLING,
+                committed=started.committed,
+                step=command.step,
+                step_state=job_pb2.STEP_STATE_CANCELLED,
+            )
+            cancelling.error.code = common_pb2.ERROR_CODE_CANCELLED
+            cancelling.error.message = str(error)
+            self._record(definition, cancelling)
+            emitted.append(cancelling)
+            if event_callback is not None:
+                event_callback(cancelling)
+            return emitted
         except Exception as error:
+            rollback_error: Exception | None = None
+            if (
+                definition.WhichOneof("spec") == "transfer"
+                and not started.committed
+            ):
+                try:
+                    if (
+                        self.client_id == definition.transfer.target_client_id
+                        and command.step in {
+                            job_pb2.JOB_STEP_KIND_TARGET_MATERIALIZE,
+                            job_pb2.JOB_STEP_KIND_QB_VERIFY,
+                        }
+                    ):
+                        self._rollback_target(definition)
+                    elif (
+                        self.client_id == definition.transfer.source_client_id
+                        and command.step == job_pb2.JOB_STEP_KIND_SOURCE_STAGE
+                    ):
+                        self._cleanup_staging(definition)
+                except Exception as compensation_error:
+                    rollback_error = compensation_error
             cleanup = (
+                (
                     command.step == job_pb2.JOB_STEP_KIND_STAGING_CLEANUP
                     and started.committed
                 )
+                or (
+                    command.step == job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK
+                    and self.store.get_job_artifact(
+                        definition.job_id, "qb-entry-removed"
+                    ) is not None
+                )
+            )
             failed = self._event(
                 definition,
-                sequence=started.sequence + 1,
+                sequence=cursor.sequence + 1,
-                revision=started.job_revision + 1,
+                revision=cursor.job_revision + 1,
                 event_type=(
                     control_pb2.JOB_EVENT_TYPE_CLEANUP_REQUIRED
                     if cleanup else control_pb2.JOB_EVENT_TYPE_FAILED
@@ -139,11 +267,23 @@ class ClientJobExecutor:
             failed.error.code = _job_error_code(error)
             failed.error.message = str(error) or type(error).__name__
             failed.error.retryable = cleanup
+            if rollback_error is not None:
+                failed.error.code = (
+                    common_pb2.ERROR_CODE_MANUAL_INTERVENTION_REQUIRED
+                )
+                failed.error.message = (
+                    f"{failed.error.message}; compensation also failed: "
+                    f"{rollback_error}"
+                )
+                failed.error.retryable = False
             self._record(definition, failed)
-            return [started, failed]
+            emitted.append(failed)
-        sequence = started.sequence + 1
+            if event_callback is not None:
-        revision = started.job_revision + 1
+                event_callback(failed)
-        committed = started.committed
+            return emitted
+        sequence = cursor.sequence + 1
+        revision = cursor.job_revision + 1
+        committed = cursor.committed
         event_type = control_pb2.JOB_EVENT_TYPE_STEP_SUCCEEDED
         state = job_pb2.JOB_STATE_RUNNING
         if command.step == job_pb2.JOB_STEP_KIND_QB_VERIFY:
@@ -152,6 +292,10 @@ class ClientJobExecutor:
         elif command.step == job_pb2.JOB_STEP_KIND_STAGING_CLEANUP:
             event_type = control_pb2.JOB_EVENT_TYPE_SUCCEEDED
             state = job_pb2.JOB_STATE_SUCCEEDED
+        elif command.step == job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK:
+            committed = True
+            event_type = control_pb2.JOB_EVENT_TYPE_SUCCEEDED
+            state = job_pb2.JOB_STATE_SUCCEEDED
         succeeded = self._event(
             definition,
             sequence=sequence,
@@ -165,24 +309,91 @@ class ClientJobExecutor:
         if result is not None:
             succeeded.observed_placement.CopyFrom(result)
         self._record(definition, succeeded)
-        return [started, succeeded]
+        emitted.append(succeeded)
+        if event_callback is not None:
+            event_callback(succeeded)
+        return emitted
+
+    def cancel(
+        self, command: control_pb2.CancelJobCommand
+    ) -> list[control_pb2.JobEvent]:
+        definition = self._definition(command.job_id)
+        self.request_cancel(command.job_id)
+        replay = self._replay(
+            command.job_id, command.expected_last_event_sequence
+        )
+        if replay:
+            return replay
+        committed = self._committed(command.job_id)
+        if definition.WhichOneof("spec") == "eviction":
+            removed = self.store.get_job_artifact(
+                definition.job_id, "qb-entry-removed"
+            )
+            placement = None
+            if removed is not None:
+                placement = self._execute_eviction_step(
+                    definition, job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK
+                )
+                committed = True
+            event_type = control_pb2.JOB_EVENT_TYPE_CANCELLED
+            state = job_pb2.JOB_STATE_CANCELLED
+        elif committed:
+            if definition.transfer.source_client_id == self.client_id:
+                self._cleanup_staging(definition)
+            event_type = control_pb2.JOB_EVENT_TYPE_CANCELLED
+            state = job_pb2.JOB_STATE_CANCELLED
+            placement = None
+        elif definition.transfer.source_client_id == self.client_id:
+            self._cleanup_staging(definition)
+            temporary = self._metainfo_path(definition.job_id)
+            if temporary.exists():
+                temporary.unlink()
+            event_type = control_pb2.JOB_EVENT_TYPE_ROLLBACK_SUCCEEDED
+            state = job_pb2.JOB_STATE_ROLLING_BACK
+            placement = None
+        else:
+            self._rollback_target(definition)
+            self._cleanup_staging(definition)
+            event_type = control_pb2.JOB_EVENT_TYPE_CANCELLED
+            state = job_pb2.JOB_STATE_CANCELLED
+            placement = None
+        event = self._event(
+            definition,
+            sequence=command.expected_last_event_sequence + 1,
+            revision=command.expected_job_revision + 1,
+            event_type=event_type,
+            state=state,
+            committed=committed,
+            step=job_pb2.JOB_STEP_KIND_ROLLBACK,
+            step_state=job_pb2.STEP_STATE_SUCCEEDED,
+        )
+        if placement is not None:
+            event.observed_placement.CopyFrom(placement)
+        self._record(definition, event)
+        return [event]
 
     def _execute_step(
-        self, definition: job_pb2.JobDefinition, step: int
+        self,
+        definition: job_pb2.JobDefinition,
+        step: int,
+        progress: Callable[[float, int, int, str], None],
     ):
+        self._raise_if_cancelled(definition.job_id)
+        if definition.WhichOneof("spec") == "eviction":
+            return self._execute_eviction_step(definition, step)
         if step == job_pb2.JOB_STEP_KIND_SOURCE_STAGE:
-            self._source_stage(definition)
+            self._source_stage(definition, progress)
             return None
         if step == job_pb2.JOB_STEP_KIND_SYNCTHING_TRANSFER:
-            self._wait_for_syncthing(definition)
+            self._wait_for_syncthing(definition, progress)
             return None
         if step == job_pb2.JOB_STEP_KIND_TARGET_MATERIALIZE:
-            self._target_materialize(definition)
+            self._target_materialize(definition, progress)
             return None
         if step == job_pb2.JOB_STEP_KIND_QB_VERIFY:
-            return self._qb_verify(definition)
+            return self._qb_verify(definition, progress)
         if step == job_pb2.JOB_STEP_KIND_STAGING_CLEANUP:
-            cleanup_transfer(self._job_directory(definition))
+            self._cleanup_staging(definition)
             temporary_metainfo = self._metainfo_path(definition.job_id)
             if temporary_metainfo.exists():
                 temporary_metainfo.unlink()
@@ -197,10 +408,86 @@ class ClientJobExecutor:
             return None
         raise JobExecutionError("job step is unsupported")
 
-    def _source_stage(self, definition: job_pb2.JobDefinition) -> None:
+    def _execute_eviction_step(
+        self, definition: job_pb2.JobDefinition, step: int
+    ):
+        if definition.eviction.cache_client_id != self.client_id:
+            raise JobExecutionError("eviction step was sent to the wrong client")
+        info_hash = _info_hash(definition)
+        if step == job_pb2.JOB_STEP_KIND_VERIFY_ARCHIVE_COVERAGE:
+            covered = {
+                index
+                for proof in definition.eviction.archive_coverage
+                for index in _selection_indices(proof.covered_files)
+            }
+            requested = set(
+                _selection_indices(definition.eviction.files_to_evict)
+            )
+            if not requested or requested - covered:
+                raise JobExecutionError(
+                    "archive coverage no longer covers the cache selection"
+                )
+            resource = self._resource(definition)
+            current = _resource_fingerprint(resource, self.client_id)
+            if not _fingerprint_matches(
+                current, definition.eviction.cache_fingerprint
+            ):
+                raise JobExecutionError(
+                    "cache resource changed after eviction confirmation"
+                )
+            verify_and_snapshot(
+                job_id=definition.job_id,
+                resource=resource,
+                selected_indices=requested,
+                qb_root=self.qb_root,
+                store=self.store,
+            )
+            return None
+        if step == job_pb2.JOB_STEP_KIND_QB_REMOVE_ENTRY:
+            remove_qb_entry(
+                job_id=definition.job_id,
+                torrent_hash=info_hash,
+                qbittorrent=self.qbittorrent,
+                store=self.store,
+            )
+            return None
+        if step == job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK:
+            safe_unlink(
+                job_id=definition.job_id,
+                qb_root=self.qb_root,
+                qbittorrent=self.qbittorrent,
+                store=self.store,
+            )
+            placement = resource_pb2.Placement(
+                client_id=self.client_id,
+                state=resource_pb2.PLACEMENT_STATE_ABSENT,
+                generation=definition.eviction.cache_placement_generation + 1,
+                created_by_job_id=definition.job_id,
+            )
+            placement.resource_id.CopyFrom(definition.resource_id)
+            placement.fingerprint.CopyFrom(
+                definition.eviction.cache_fingerprint
+            )
+            placement.verified_at.GetCurrentTime()
+            return placement
+        raise JobExecutionError("eviction job step is unsupported")
+
+    def _source_stage(
+        self,
+        definition: job_pb2.JobDefinition,
+        progress: Callable[[float, int, int, str], None],
+    ) -> None:
         if definition.transfer.source_client_id != self.client_id:
             raise JobExecutionError("source stage was sent to the wrong client")
         resource = self._resource(definition)
+        if not _fingerprint_matches(
+            _resource_fingerprint(resource, self.client_id),
+            definition.transfer.source_fingerprint,
+        ):
+            raise JobExecutionError(
+                "source resource changed after job confirmation"
+            )
+        self._reject_unsafe_partfile(definition, resource)
         indices = _selection_indices(definition.transfer.transfer_delta_files)
         by_index = {item.file_index: item for item in resource.files}
         if not indices or any(index not in by_index for index in indices):
@@ -254,6 +541,11 @@ class ClientJobExecutor:
             sha256_hex=hashlib.sha256(resource.metainfo_bytes).hexdigest(),
         )
         del artifact
+        self._require_space(
+            self.route_path(definition.transfer.route_id),
+            definition.transfer.transfer_delta_logical_bytes
+            + len(resource.metainfo_bytes),
+        )
         stage_transfer(
             manifest,
             source_root=self.qb_root,
@@ -261,18 +553,43 @@ class ClientJobExecutor:
             store=self.store,
             artifact_sources={"metainfo/source.torrent": metainfo_path},
             sparse_supported=self.sparse_supported,
+            cancel_check=lambda: self._raise_if_cancelled(
+                definition.job_id
+            ),
+            progress=lambda completed, total: progress(
+                completed / total if total else 1,
+                completed,
+                total,
+                f"staged {completed} of {total} bytes",
+            ),
         )
         self._observer(definition).rescan()
 
-    def _wait_for_syncthing(self, definition: job_pb2.JobDefinition) -> None:
+    def _wait_for_syncthing(
+        self,
+        definition: job_pb2.JobDefinition,
+        progress: Callable[[float, int, int, str], None],
+    ) -> None:
         if definition.transfer.target_client_id != self.client_id:
             raise JobExecutionError(
                 "Syncthing completion was sent to the wrong client"
             )
         observer = self._observer(definition)
+        last_reported: tuple[int, int] | None = None
         while True:
+            self._raise_if_cancelled(definition.job_id)
             try:
-                if observer.status().complete:
+                status = observer.status()
+                sample = (status.bytes_complete, status.bytes_total)
+                if sample != last_reported:
+                    progress(
+                        status.fraction_complete,
+                        status.bytes_complete,
+                        status.bytes_total,
+                        f"{status.needed_items} Syncthing items still needed",
+                    )
+                    last_reported = sample
+                if status.complete:
                     return
             except (FileNotFoundError, TransferIntegrityError):
                 # Syncthing can expose the per-job directory before the
@@ -280,20 +597,73 @@ class ClientJobExecutor:
                 pass
             time.sleep(self.poll_interval)
 
-    def _target_materialize(self, definition: job_pb2.JobDefinition) -> None:
+    def _target_materialize(
+        self,
+        definition: job_pb2.JobDefinition,
+        progress: Callable[[float, int, int, str], None],
+    ) -> None:
         if definition.transfer.target_client_id != self.client_id:
             raise JobExecutionError(
                 "target materialization was sent to the wrong client"
             )
         published = load_published_transfer(self._job_directory(definition))
+        self._require_space(
+            self.qb_root,
+            definition.transfer.transfer_delta_logical_bytes,
+        )
+        info_hash = _info_hash(definition)
+        resource = self.qbittorrent.get_resource(info_hash)
+        if resource is not None:
+            self._reject_unsafe_partfile(definition, resource)
+        if definition.transfer.HasField("target_baseline_fingerprint"):
+            if resource is None or not _fingerprint_matches(
+                _resource_fingerprint(resource, self.client_id),
+                definition.transfer.target_baseline_fingerprint,
+            ):
+                raise JobExecutionError(
+                    "target resource changed after job confirmation"
+                )
+        elif resource is not None:
+            raise JobExecutionError(
+                "target resource appeared after job confirmation"
+            )
+        baseline = {
+            "present": resource is not None,
+            "selected_file_indices": (
+                _selection_indices(resource.summary.selected_files)
+                if resource is not None else []
+            ),
+            "stopped": (
+                resource is not None
+                and resource.summary.runtime_state
+                == resource_pb2.TORRENT_RUNTIME_STATE_STOPPED
+            ),
+            "total_file_count": len(published.manifest.files),
+        }
+        self.store.put_job_artifact(
+            definition.job_id, "target-baseline", baseline
+        )
         materialize_transfer(
             published,
             target_root=self.qb_root,
             store=self.store,
             sparse_supported=self.sparse_supported,
+            cancel_check=lambda: self._raise_if_cancelled(
+                definition.job_id
+            ),
+            progress=lambda completed, total: progress(
+                completed / total if total else 1,
+                completed,
+                total,
+                f"materialized {completed} of {total} bytes",
+            ),
         )
 
-    def _qb_verify(self, definition: job_pb2.JobDefinition):
+    def _qb_verify(
+        self,
+        definition: job_pb2.JobDefinition,
+        progress: Callable[[float, int, int, str], None],
+    ):
         if definition.transfer.target_client_id != self.client_id:
             raise JobExecutionError("qB verification was sent to the wrong client")
         published = load_published_transfer(self._job_directory(definition))
@@ -326,7 +696,16 @@ class ClientJobExecutor:
         )
         self.qbittorrent.set_selection(info_hash, selected, total_files)
         self.qbittorrent.recheck_and_wait(
-            info_hash, selected, timeout=30 * 60, poll_interval=self.poll_interval
+            info_hash,
+            selected,
+            timeout=self.verification_timeout,
+            poll_interval=self.poll_interval,
+            cancel_check=lambda: self._raise_if_cancelled(
+                definition.job_id
+            ),
+            progress_callback=lambda fraction: progress(
+                fraction, 0, 0, "qBittorrent stopped recheck"
+            ),
         )
         if should_start:
             self.qbittorrent.start(info_hash)
@@ -359,6 +738,102 @@ class ClientJobExecutor:
         placement.verified_at.GetCurrentTime()
         return placement
 
+    def _rollback_target(self, definition: job_pb2.JobDefinition) -> None:
+        baseline_row = self.store.get_job_artifact(
+            definition.job_id, "target-baseline"
+        )
+        if baseline_row is None:
+            compensate_materialized_files(
+                job_id=definition.job_id,
+                qb_root=self.qb_root,
+                store=self.store,
+            )
+            return
+        baseline = baseline_row["value"]
+        info_hash = _info_hash(definition)
+        present = self.qbittorrent.get_resource(info_hash)
+        if present is not None:
+            if baseline.get("present"):
+                selected = baseline.get("selected_file_indices")
+                if isinstance(selected, list) and selected:
+                    self.qbittorrent.set_selection(
+                        info_hash, selected, len(present.files)
+                    )
+                if baseline.get("stopped"):
+                    self.qbittorrent.stop(info_hash)
+                else:
+                    self.qbittorrent.start(info_hash)
+            else:
+                self.qbittorrent.delete_entry(info_hash)
+        compensate_materialized_files(
+            job_id=definition.job_id,
+            qb_root=self.qb_root,
+            store=self.store,
+        )
+
+    def _cleanup_staging(self, definition: job_pb2.JobDefinition) -> None:
+        job_directory = self._job_directory(definition)
+        try:
+            if cleanup_transfer(job_directory):
+                return
+        except FileNotFoundError:
+            pass
+        if not job_directory.is_dir():
+            return
+        compensate_materialized_files(
+            job_id=definition.job_id,
+            qb_root=job_directory,
+            store=self.store,
+        )
+        for name in ("ready.json", "manifest.json"):
+            candidate = job_directory / name
+            if candidate.is_file() and not candidate.is_symlink():
+                candidate.unlink()
+        try:
+            job_directory.rmdir()
+        except OSError:
+            pass
+
+    def _reject_unsafe_partfile(
+        self,
+        definition: job_pb2.JobDefinition,
+        resource: NormalizedResource,
+    ) -> None:
+        roots = {self.qb_root}
+        for item in resource.files:
+            candidate = self.qb_root / PurePosixPath(item.canonical_path).parts[0]
+            roots.add(candidate if candidate.is_dir() else candidate.parent)
+        hashes = {
+            value.lower() for value in (
+                definition.resource_id.info_hash_v1_hex,
+                definition.resource_id.info_hash_v2_hex,
+            ) if value
+        }
+        for root in roots:
+            if not root.is_dir():
+                continue
+            for candidate in root.glob("*.parts"):
+                lowered = candidate.name.lower()
+                if lowered == ".parts" or any(value in lowered for value in hashes):
+                    raise JobExecutionError(
+                        "resource uses a qBittorrent partfile that cannot be "
+                        "safely transferred by this client"
+                    )
+
+    def _raise_if_cancelled(self, job_id: str) -> None:
+        event = self._cancel_events.get(job_id)
+        if event is not None and event.is_set():
+            raise JobCancelled("job cancellation requested")
+
+    def _require_space(self, root: Path, required_bytes: int) -> None:
+        available = shutil.disk_usage(root).free
+        required = required_bytes + self.free_space_reserve_bytes
+        if available < required:
+            raise JobExecutionError(
+                f"insufficient free space: {available} bytes available, "
+                f"{required} bytes required including reserve"
+            )
+
     def _observer(
         self, definition: job_pb2.JobDefinition
     ) -> SyncthingTransferObserver:
@@ -455,21 +930,29 @@ class ClientJobExecutor:
             )
             event.progress.overall_fraction_complete = _overall(step, step_state)
             event.progress.display_step_number = _step_number(step)
-            event.progress.display_step_total = 5
+            event.progress.display_step_total = (
+                3
+                if definition.operation == job_pb2.JOB_OPERATION_EVICT_CACHE
+                else 5
+            )
             event.progress.last_progress_at.CopyFrom(event.occurred_at)
         return event
 
     @staticmethod
     def _validate_definition(definition: job_pb2.JobDefinition) -> None:
-        if (
+        if not definition.job_id:
-            not definition.job_id
+            raise JobExecutionError("job definition has no ID")
-            or definition.operation not in {
+        if definition.operation in {
             job_pb2.JOB_OPERATION_ARCHIVE,
             job_pb2.JOB_OPERATION_UNARCHIVE,
-            }
+        } and definition.WhichOneof("spec") == "transfer":
-            or definition.WhichOneof("spec") != "transfer"
+            return
+        if (
+            definition.operation == job_pb2.JOB_OPERATION_EVICT_CACHE
+            and definition.WhichOneof("spec") == "eviction"
         ):
-            raise JobExecutionError("transfer job definition is invalid")
+            return
+        raise JobExecutionError("job definition operation/spec is invalid")
 
 
 def _selection_indices(selection) -> list[int]:
@@ -510,12 +993,27 @@ def _step_number(step: int) -> int:
         job_pb2.JOB_STEP_KIND_TARGET_MATERIALIZE: 3,
         job_pb2.JOB_STEP_KIND_QB_VERIFY: 4,
         job_pb2.JOB_STEP_KIND_STAGING_CLEANUP: 5,
+        job_pb2.JOB_STEP_KIND_VERIFY_ARCHIVE_COVERAGE: 1,
+        job_pb2.JOB_STEP_KIND_QB_REMOVE_ENTRY: 2,
+        job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK: 3,
+        job_pb2.JOB_STEP_KIND_ROLLBACK: 1,
     }.get(step, 0)
 
 
 def _overall(step: int, state: int) -> float:
     number = _step_number(step)
-    return max(0, number - (0 if state == job_pb2.STEP_STATE_SUCCEEDED else 1)) / 5
+    total = (
+        3
+        if step in {
+            job_pb2.JOB_STEP_KIND_VERIFY_ARCHIVE_COVERAGE,
+            job_pb2.JOB_STEP_KIND_QB_REMOVE_ENTRY,
+            job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK,
+        }
+        else 5
+    )
+    return max(
+        0, number - (0 if state == job_pb2.STEP_STATE_SUCCEEDED else 1)
+    ) / total
 
 
 def decode_message_metainfo(path: Path):
@@ -543,6 +1041,38 @@ def _resource_fingerprint(
     return fingerprint
 
 
+def _info_hash(definition: job_pb2.JobDefinition) -> str:
+    value = (
+        definition.resource_id.info_hash_v1_hex
+        or definition.resource_id.info_hash_v2_hex
+    )
+    if not value:
+        raise JobExecutionError("job resource identity has no info hash")
+    return value
+
+
+def _fingerprint_matches(
+    current: resource_pb2.ResourceStateFingerprint,
+    expected: resource_pb2.ResourceStateFingerprint,
+) -> bool:
+    left = resource_pb2.ResourceStateFingerprint()
+    right = resource_pb2.ResourceStateFingerprint()
+    left.CopyFrom(current)
+    right.CopyFrom(expected)
+    left.ClearField("observed_at")
+    right.ClearField("observed_at")
+    for value in (left, right):
+        if value.runtime_state in {
+            resource_pb2.TORRENT_RUNTIME_STATE_QUEUED,
+            resource_pb2.TORRENT_RUNTIME_STATE_SEEDING,
+            resource_pb2.TORRENT_RUNTIME_STATE_STALLED,
+        }:
+            value.runtime_state = (
+                resource_pb2.TORRENT_RUNTIME_STATE_SEEDING
+            )
+    return left == right
+
+
 def _job_error_code(error: Exception) -> int:
     if isinstance(error, QBittorrentDownloadAttempt):
         return common_pb2.ERROR_CODE_MANUAL_INTERVENTION_REQUIRED
@@ -554,4 +1084,6 @@ def _job_error_code(error: Exception) -> int:
         return common_pb2.ERROR_CODE_PRECONDITION_FAILED
     if isinstance(error, TransferError):
         return common_pb2.ERROR_CODE_PATH_CONFLICT
+    if isinstance(error, EvictionError):
+        return common_pb2.ERROR_CODE_PRECONDITION_FAILED
     return common_pb2.ERROR_CODE_INTERNAL

src/archive_clients/qbittorrent.py

@@ -10,6 +10,7 @@ from dataclasses import dataclass
 from http.cookiejar import CookieJar
 from pathlib import PurePosixPath
 from typing import Any
+from typing import Callable
 from urllib import error, parse, request
 
 from archive_clients.config import ServiceConfig
@@ -259,6 +260,8 @@ class QBittorrentReader:
         *,
         timeout: float,
         poll_interval: float = 1,
+        cancel_check: Callable[[], None] | None = None,
+        progress_callback: Callable[[float], None] | None = None,
     ) -> RecheckResult:
         selected = tuple(sorted(set(selected_file_indices)))
         if not selected:
@@ -271,7 +274,9 @@ class QBittorrentReader:
             "/api/v2/torrents/recheck", {"hashes": torrent_hash}
         )
         deadline = time.monotonic() + timeout
+        cancel_check = cancel_check or (lambda: None)
         while True:
+            cancel_check()
             record = self._torrent_record(torrent_hash)
             state = record.get("state")
             if not isinstance(state, str):
@@ -295,6 +300,13 @@ class QBittorrentReader:
                 item.get("index"): item.get("progress")
                 for item in files
             }
+            if progress_callback is not None:
+                progress_callback(
+                    sum(
+                        max(0.0, min(float(progress.get(index, 0)), 1.0))
+                        for index in selected
+                    ) / len(selected)
+                )
             if all(
                 isinstance(progress.get(index), (int, float))
                 and float(progress[index]) >= 1

src/archive_clients/state.py

@@ -11,7 +11,7 @@ from dataclasses import dataclass
 from pathlib import Path
 
 
-SCHEMA_VERSION = 2
+SCHEMA_VERSION = 3
 
 
 class CommandConflict(RuntimeError):
@@ -58,6 +58,10 @@ class ClientStore:
             if version == 1:
                 connection.executescript(_SCHEMA_V2)
                 connection.execute("PRAGMA user_version = 2")
+                version = 2
+            if version == 2:
+                connection.executescript(_SCHEMA_V3)
+                connection.execute("PRAGMA user_version = 3")
             if connection.execute("PRAGMA foreign_key_check").fetchall():
                 raise RuntimeError("client database foreign-key check failed")
         if not existed:
@@ -268,6 +272,57 @@ class ClientStore:
             ).fetchall()
         return [dict(row) for row in rows]
 
+    def put_job_artifact(
+        self, job_id: str, kind: str, value: dict[str, object]
+    ) -> dict[str, object]:
+        encoded = _canonical(value)
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            existing = connection.execute(
+                """
+                SELECT value_json FROM job_artifacts
+                WHERE job_id = ? AND kind = ?
+                """,
+                (job_id, kind),
+            ).fetchone()
+            if existing is not None and existing["value_json"] != encoded:
+                raise JobConflict(f"job artifact {kind} is immutable")
+            connection.execute(
+                """
+                INSERT INTO job_artifacts (job_id, kind, value_json)
+                VALUES (?, ?, ?)
+                ON CONFLICT(job_id, kind) DO NOTHING
+                """,
+                (job_id, kind, encoded),
+            )
+            row = connection.execute(
+                """
+                SELECT job_id, kind, value_json FROM job_artifacts
+                WHERE job_id = ? AND kind = ?
+                """,
+                (job_id, kind),
+            ).fetchone()
+        result = dict(row)
+        result["value"] = json.loads(str(result.pop("value_json")))
+        return result
+
+    def get_job_artifact(
+        self, job_id: str, kind: str
+    ) -> dict[str, object] | None:
+        with self._connect() as connection:
+            row = connection.execute(
+                """
+                SELECT job_id, kind, value_json FROM job_artifacts
+                WHERE job_id = ? AND kind = ?
+                """,
+                (job_id, kind),
+            ).fetchone()
+        if row is None:
+            return None
+        result = dict(row)
+        result["value"] = json.loads(str(result.pop("value_json")))
+        return result
+
     def record_job_event(
         self,
         *,
@@ -567,3 +622,13 @@ CREATE TABLE route_attempt_updates (
     PRIMARY KEY(command_id, sequence)
 );
 """
+
+
+_SCHEMA_V3 = """
+CREATE TABLE job_artifacts (
+    job_id TEXT NOT NULL REFERENCES jobs(job_id),
+    kind TEXT NOT NULL,
+    value_json TEXT NOT NULL,
+    PRIMARY KEY(job_id, kind)
+);
+"""

src/archive_clients/transfer.py

@@ -11,7 +11,7 @@ import stat
 import uuid
 from dataclasses import dataclass
 from pathlib import Path, PurePosixPath
-from typing import Mapping
+from typing import Callable, Mapping
 
 from google.protobuf import json_format
 
@@ -79,6 +79,7 @@ class FileMaterializer:
         allow_hardlink: bool = True,
         allow_reflink: bool = True,
         copy_chunk_bytes: int = _COPY_CHUNK_BYTES,
+        cancel_check: Callable[[], None] | None = None,
     ):
         if copy_chunk_bytes < 1:
             raise ValueError("copy_chunk_bytes must be positive")
@@ -86,6 +87,7 @@ class FileMaterializer:
         self.allow_hardlink = allow_hardlink
         self.allow_reflink = allow_reflink
         self.copy_chunk_bytes = copy_chunk_bytes
+        self.cancel_check = cancel_check or (lambda: None)
 
     def materialize(
         self,
@@ -100,6 +102,7 @@ class FileMaterializer:
         allow_preexisting_reuse: bool,
         sparse_supported: bool,
     ) -> MaterializedFile:
+        self.cancel_check()
         source_relative = _relative_path(source_relative_path)
         destination_relative = _relative_path(destination_relative_path)
         source = _existing_regular_file(source_root, source_relative)
@@ -184,6 +187,7 @@ class FileMaterializer:
                 destination,
                 operation_id,
                 self.copy_chunk_bytes,
+                self.cancel_check,
             )
             method = transfer_pb2.MATERIALIZATION_METHOD_COPY
 
@@ -233,6 +237,8 @@ def stage_transfer(
     allow_hardlink: bool = True,
     allow_reflink: bool = True,
     sparse_supported: bool = True,
+    cancel_check: Callable[[], None] | None = None,
+    progress: Callable[[int, int], None] | None = None,
 ) -> PublishedTransfer:
     """Publish a complete isolated transfer namespace and ready marker."""
 
@@ -244,7 +250,12 @@ def stage_transfer(
         store,
         allow_hardlink=allow_hardlink,
         allow_reflink=allow_reflink,
+        cancel_check=cancel_check,
     )
+    total_bytes = sum(entry.logical_bytes for entry in staged.files) + sum(
+        artifact.logical_bytes for artifact in staged.artifacts
+    )
+    completed_bytes = 0
     for entry in staged.files:
         result = materializer.materialize(
             job_id=staged.job_id,
@@ -263,6 +274,9 @@ def stage_transfer(
         entry.allocated_bytes = result.allocated_bytes
         entry.sparse = result.sparse
         entry.source_staging_method = result.method
+        completed_bytes += entry.logical_bytes
+        if progress is not None:
+            progress(completed_bytes, total_bytes)
 
     artifact_sources = artifact_sources or {}
     for artifact in staged.artifacts:
@@ -287,6 +301,9 @@ def stage_transfer(
         )
         artifact.allocated_bytes = result.allocated_bytes
         artifact.sparse = result.sparse
+        completed_bytes += artifact.logical_bytes
+        if progress is not None:
+            progress(completed_bytes, total_bytes)
         digest = _sha256_file(
             _existing_regular_file(
                 job_directory, _relative_path(artifact.payload_relative_path)
@@ -356,6 +373,8 @@ def materialize_transfer(
     allow_hardlink: bool = True,
     allow_reflink: bool = True,
     sparse_supported: bool = True,
+    cancel_check: Callable[[], None] | None = None,
+    progress: Callable[[int, int], None] | None = None,
 ) -> transfer_pb2.TransferManifest:
     """Materialize a verified published payload into a target content root."""
 
@@ -366,7 +385,10 @@ def materialize_transfer(
         store,
         allow_hardlink=allow_hardlink,
         allow_reflink=allow_reflink,
+        cancel_check=cancel_check,
     )
+    total_bytes = sum(entry.logical_bytes for entry in result_manifest.files)
+    completed_bytes = 0
     for entry in result_manifest.files:
         result = materializer.materialize(
             job_id=result_manifest.job_id,
@@ -384,6 +406,9 @@ def materialize_transfer(
         )
         entry.target_method = result.method
         entry.target_preexisted = result.destination_preexisted
+        completed_bytes += entry.logical_bytes
+        if progress is not None:
+            progress(completed_bytes, total_bytes)
     return result_manifest
 
 
@@ -455,20 +480,35 @@ def _validate_manifest(manifest: transfer_pb2.TransferManifest) -> None:
         raise TransferIntegrityError("manifest creation timestamp is required")
     indices: set[int] = set()
     payload_paths: set[str] = set()
+    folded_payload_paths: set[str] = set()
+    target_paths: set[str] = set()
+    folded_target_paths: set[str] = set()
     for entry in manifest.files:
         if entry.file_index in indices:
             raise TransferIntegrityError("manifest contains duplicate file indices")
         indices.add(entry.file_index)
         payload = _relative_path(entry.payload_relative_path).as_posix()
-        _relative_path(entry.target_canonical_path)
+        target = _relative_path(entry.target_canonical_path).as_posix()
         if payload in payload_paths:
             raise TransferIntegrityError("manifest contains duplicate payload paths")
+        if payload.casefold() in folded_payload_paths:
+            raise TransferIntegrityError("manifest contains case-colliding payload paths")
+        if target in target_paths:
+            raise TransferIntegrityError("manifest contains duplicate target paths")
+        if target.casefold() in folded_target_paths:
+            raise TransferIntegrityError("manifest contains case-colliding target paths")
         payload_paths.add(payload)
+        folded_payload_paths.add(payload.casefold())
+        target_paths.add(target)
+        folded_target_paths.add(target.casefold())
     for artifact in manifest.artifacts:
         payload = _relative_path(artifact.payload_relative_path).as_posix()
         if payload in payload_paths:
             raise TransferIntegrityError("manifest contains duplicate payload paths")
+        if payload.casefold() in folded_payload_paths:
+            raise TransferIntegrityError("manifest contains case-colliding payload paths")
         payload_paths.add(payload)
+        folded_payload_paths.add(payload.casefold())
 
 
 def _relative_path(value: str) -> PurePosixPath:
@@ -569,6 +609,7 @@ def _sparse_copy(
     destination: Path,
     operation_id: str,
     chunk_bytes: int,
+    cancel_check: Callable[[], None],
 ) -> None:
     temporary = _temporary_path(destination, operation_id)
     source_fd = os.open(source, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
@@ -580,7 +621,13 @@ def _sparse_copy(
             os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
             stat.S_IMODE(source_stat.st_mode),
         )
-        _copy_extents(source_fd, destination_fd, source_stat.st_size, chunk_bytes)
+        _copy_extents(
+            source_fd,
+            destination_fd,
+            source_stat.st_size,
+            chunk_bytes,
+            cancel_check,
+        )
         os.ftruncate(destination_fd, source_stat.st_size)
         os.fsync(destination_fd)
         _publish_temporary(temporary, destination)
@@ -596,6 +643,7 @@ def _copy_extents(
     destination_fd: int,
     size: int,
     chunk_bytes: int,
+    cancel_check: Callable[[], None],
 ) -> None:
     if size == 0:
         return
@@ -613,6 +661,7 @@ def _copy_extents(
             os.lseek(destination_fd, data_offset, os.SEEK_SET)
             remaining = min(hole_offset, size) - data_offset
             while remaining:
+                cancel_check()
                 data = os.read(source_fd, min(chunk_bytes, remaining))
                 if not data:
                     raise TransferIntegrityError("source ended during sparse copy")
@@ -626,6 +675,7 @@ def _copy_extents(
         os.lseek(destination_fd, 0, os.SEEK_SET)
         remaining = size
         while remaining:
+            cancel_check()
             data = os.read(source_fd, min(chunk_bytes, remaining))
             if not data:
                 raise TransferIntegrityError("source ended during copy")

tests/test_daemon.py

@@ -22,6 +22,51 @@ from archive_control.v1 import (
 
 
 class DaemonTransportTests(unittest.IsolatedAsyncioTestCase):
+    async def test_eviction_assignment_and_steps_are_admitted(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            token = root / "token"
+            token.write_text("shared-secret", encoding="utf-8")
+            os.chmod(token, 0o600)
+            service = ServiceConfig(
+                "http://local", PurePosixPath("/api"), root
+            )
+            config = ClientConfig(
+                "cache-1", "Cache 1", "cache", "ws://control", token,
+                root / "state.db", root / "backups", service, service,
+            )
+            probe = FilesystemProbe(root, True, True, True, True, True)
+            daemon = ArchiveClientDaemon(config, [probe, probe], [])
+            daemon.jobs = Mock()
+
+            assign = control_pb2.Command(command_id=str(uuid4()))
+            definition = assign.assign_job.job
+            definition.job_id = str(uuid4())
+            definition.eviction.cache_client_id = "cache-1"
+            acknowledgement = daemon._initial_acknowledgement(
+                assign, set(), False
+            )
+            self.assertEqual(
+                acknowledgement.status,
+                control_pb2.COMMAND_ACK_STATUS_ACCEPTED,
+            )
+
+            for step_kind in (
+                job_pb2.JOB_STEP_KIND_VERIFY_ARCHIVE_COVERAGE,
+                job_pb2.JOB_STEP_KIND_QB_REMOVE_ENTRY,
+                job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK,
+            ):
+                execute = control_pb2.Command(command_id=str(uuid4()))
+                execute.execute_step.job_id = definition.job_id
+                execute.execute_step.step = step_kind
+                acknowledgement = daemon._initial_acknowledgement(
+                    execute, set(), False
+                )
+                self.assertEqual(
+                    acknowledgement.status,
+                    control_pb2.COMMAND_ACK_STATUS_ACCEPTED,
+                )
+
     async def test_ensure_route_is_durable_and_duplicate_replays_updates(self):
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory)

tests/test_eviction.py

@@ -0,0 +1,181 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from archive_clients.eviction import (
+    remove_qb_entry,
+    safe_unlink,
+    verify_and_snapshot,
+)
+from archive_clients.resources import NormalizedResource
+from archive_clients.state import ClientStore
+from archive_control.v1 import resource_pb2
+
+
+def normalized(torrent_hash, paths):
+    summary = resource_pb2.ResourceSummary(
+        qb_torrent_id=torrent_hash,
+        display_name="resource",
+        total_file_count=len(paths),
+        canonical_paths=True,
+    )
+    summary.resource_id.info_hash_v1_hex = torrent_hash
+    files = []
+    for index, path in enumerate(paths):
+        size = len(path) + 10
+        files.append(resource_pb2.TorrentFile(
+            file_index=index,
+            canonical_path=path,
+            logical_bytes=size,
+            completed_bytes=size,
+            selected=True,
+        ))
+        summary.selected_files.ranges.add(first=index, last=index)
+        summary.selected_complete_files.ranges.add(first=index, last=index)
+    return NormalizedResource(summary, tuple(files), None)
+
+
+class FakeQB:
+    def __init__(self, evicted, remaining):
+        self.evicted = evicted
+        self.remaining = remaining
+        self.deleted = []
+
+    def get_resource(self, torrent_hash):
+        return (
+            self.evicted
+            if self.evicted
+            and self.evicted.summary.qb_torrent_id == torrent_hash
+            else None
+        )
+
+    def delete_entry(self, torrent_hash):
+        self.deleted.append(torrent_hash)
+        self.evicted = None
+
+    def list_resources(self):
+        return list(self.remaining)
+
+
+class EvictionTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name) / "qb"
+        self.root.mkdir()
+        self.store = ClientStore(Path(self.temp.name) / "state.db")
+        self.store.initialize()
+        self.store.save_job(
+            "job-1", '{"jobId":"job-1"}', "JOB_STATE_RUNNING", 1, 0, False
+        )
+
+    def tearDown(self):
+        self.temp.cleanup()
+
+    def test_shared_paths_and_unknown_files_survive(self):
+        evicted = normalized("a" * 40, ["tree/shared.bin", "tree/owned.bin"])
+        other = normalized("b" * 40, ["tree/shared.bin"])
+        for item in evicted.files:
+            path = self.root / item.canonical_path
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_bytes(b"x" * item.logical_bytes)
+        unknown = self.root / "tree" / "from-another-app.txt"
+        unknown.write_text("keep")
+        qb = FakeQB(evicted, [other])
+
+        verify_and_snapshot(
+            job_id="job-1",
+            resource=evicted,
+            selected_indices=[0, 1],
+            qb_root=self.root,
+            store=self.store,
+        )
+        remove_qb_entry(
+            job_id="job-1",
+            torrent_hash="a" * 40,
+            qbittorrent=qb,
+            store=self.store,
+        )
+        result = safe_unlink(
+            job_id="job-1",
+            qb_root=self.root,
+            qbittorrent=qb,
+            store=self.store,
+        )
+        self.assertTrue((self.root / "tree/shared.bin").exists())
+        self.assertFalse((self.root / "tree/owned.bin").exists())
+        self.assertTrue(unknown.exists())
+        self.assertTrue((self.root / "tree").is_dir())
+        self.assertEqual(qb.deleted, ["a" * 40])
+        self.assertEqual(
+            result["retained_files"],
+            [{"path": "tree/shared.bin", "reason": "shared"}],
+        )
+
+    def test_changed_inode_is_not_unlinked(self):
+        evicted = normalized("a" * 40, ["tree/owned.bin"])
+        path = self.root / "tree/owned.bin"
+        path.parent.mkdir(parents=True)
+        path.write_bytes(b"x" * evicted.files[0].logical_bytes)
+        qb = FakeQB(evicted, [])
+        verify_and_snapshot(
+            job_id="job-1",
+            resource=evicted,
+            selected_indices=[0],
+            qb_root=self.root,
+            store=self.store,
+        )
+        path.unlink()
+        path.write_bytes(b"y" * evicted.files[0].logical_bytes)
+        remove_qb_entry(
+            job_id="job-1",
+            torrent_hash="a" * 40,
+            qbittorrent=qb,
+            store=self.store,
+        )
+        result = safe_unlink(
+            job_id="job-1",
+            qb_root=self.root,
+            qbittorrent=qb,
+            store=self.store,
+        )
+        self.assertTrue(path.exists())
+        self.assertEqual(
+            result["retained_files"][0]["reason"], "identity-changed"
+        )
+
+    def test_retry_is_idempotent(self):
+        evicted = normalized("a" * 40, ["owned.bin"])
+        path = self.root / "owned.bin"
+        path.write_bytes(b"x" * evicted.files[0].logical_bytes)
+        qb = FakeQB(evicted, [])
+        verify_and_snapshot(
+            job_id="job-1",
+            resource=evicted,
+            selected_indices=[0],
+            qb_root=self.root,
+            store=self.store,
+        )
+        remove_qb_entry(
+            job_id="job-1",
+            torrent_hash="a" * 40,
+            qbittorrent=qb,
+            store=self.store,
+        )
+        first = safe_unlink(
+            job_id="job-1",
+            qb_root=self.root,
+            qbittorrent=qb,
+            store=self.store,
+        )
+        second = safe_unlink(
+            job_id="job-1",
+            qb_root=self.root,
+            qbittorrent=qb,
+            store=self.store,
+        )
+        self.assertEqual(first, second)
+        self.assertEqual(qb.deleted, ["a" * 40])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_jobs.py

@@ -6,7 +6,11 @@ from unittest.mock import Mock, patch
 from uuid import uuid4
 
 from archive_clients.bencode import encode
-from archive_clients.jobs import ClientJobExecutor, JobExecutionError
+from archive_clients.jobs import (
+    ClientJobExecutor,
+    JobExecutionError,
+    _resource_fingerprint,
+)
 from archive_clients.resources import normalize_resource
 from archive_clients.state import ClientStore
 from archive_control.v1 import control_pb2, job_pb2
@@ -90,6 +94,9 @@ class ClientJobHappyPathTests(unittest.TestCase):
         definition.created_at.GetCurrentTime()
         definition.transfer.requested_files.ranges.add(first=0, last=0)
         definition.transfer.transfer_delta_files.ranges.add(first=0, last=0)
+        definition.transfer.source_fingerprint.CopyFrom(
+            _resource_fingerprint(resource, source_id)
+        )
 
         source_store = ClientStore(root / "source.db")
         target_store = ClientStore(root / "target.db")
@@ -98,7 +105,7 @@ class ClientJobHappyPathTests(unittest.TestCase):
         source_qb = Mock()
         source_qb.get_resource.return_value = resource
         target_qb = Mock()
-        target_qb.get_resource.side_effect = [None, resource]
+        target_qb.get_resource.side_effect = [None, None, resource]
         syncthing = CompleteSyncthing()
         source = ClientJobExecutor(
             client_id=source_id,
@@ -156,7 +163,10 @@ class ClientJobHappyPathTests(unittest.TestCase):
             ))
             self.assertEqual(
                 [event.sequence for event in events],
-                [cursor_sequence + 1, cursor_sequence + 2],
+                list(range(
+                    cursor_sequence + 1,
+                    cursor_sequence + len(events) + 1,
+                )),
             )
             cursor_sequence = events[-1].sequence
             cursor_revision = events[-1].job_revision
@@ -250,6 +260,54 @@ class ClientJobHappyPathTests(unittest.TestCase):
             )
             self.assertEqual(store.list_active_job_cursors(), [])
 
+    def test_active_source_cancellation_reports_durable_rollback(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            store = ClientStore(root / "client.db")
+            store.initialize()
+            definition = job_pb2.JobDefinition(
+                job_id=str(uuid4()),
+                idempotency_key=str(uuid4()),
+                operation=job_pb2.JOB_OPERATION_ARCHIVE,
+                resource_display_name="fixture",
+                transfer={
+                    "source_client_id": "cache-1",
+                    "target_client_id": "archive-1",
+                    "route_id": "route-1",
+                },
+            )
+            definition.resource_id.info_hash_v1_hex = "a" * 40
+            definition.created_at.GetCurrentTime()
+            executor = ClientJobExecutor(
+                client_id="cache-1",
+                qbittorrent=Mock(),
+                store=store,
+                qb_root=root,
+                qb_api_root=Path("/downloads"),
+                route_path=lambda _: root / "route",
+                syncthing_transport=Mock(),
+                sparse_supported=True,
+                poll_interval=0,
+            )
+            executor.assign(control_pb2.AssignJobCommand(
+                job=definition,
+                expected_job_revision=1,
+                expected_last_event_sequence=0,
+            ))
+            events = executor.cancel(control_pb2.CancelJobCommand(
+                job_id=definition.job_id,
+                expected_job_revision=2,
+                expected_last_event_sequence=1,
+                reason="test",
+            ))
+            self.assertEqual(
+                events[-1].type,
+                control_pb2.JOB_EVENT_TYPE_ROLLBACK_SUCCEEDED,
+            )
+            self.assertEqual(
+                events[-1].state, job_pb2.JOB_STATE_ROLLING_BACK
+            )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_state.py

@@ -120,6 +120,28 @@ class ClientStoreTests(unittest.TestCase):
                     "operation-1", '{"method":2}'
                 )
 
+    def test_job_artifacts_are_immutable_and_survive_restart(self):
+        with tempfile.TemporaryDirectory() as directory:
+            database = Path(directory) / "state.db"
+            store = ClientStore(database)
+            store.initialize()
+            store.save_job(
+                "job-1", '{"jobId":"job-1"}', "JOB_STATE_RUNNING",
+                1, 0, False,
+            )
+            store.put_job_artifact(
+                "job-1", "baseline", {"selected": [1, 3]}
+            )
+            reopened = ClientStore(database)
+            self.assertEqual(
+                reopened.get_job_artifact("job-1", "baseline")["value"],
+                {"selected": [1, 3]},
+            )
+            with self.assertRaises(JobConflict):
+                reopened.put_job_artifact(
+                    "job-1", "baseline", {"selected": [2]}
+                )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_transfer.py

@@ -9,6 +9,7 @@ from uuid import uuid4
 from archive_clients.state import ClientStore
 from archive_clients.transfer import (
     FileMaterializer,
+    TransferIntegrityError,
     canonical_message_json,
     load_published_transfer,
     materialize_transfer,
@@ -212,6 +213,22 @@ class TransferHappyPathTests(unittest.TestCase):
         self.assertEqual(first, second)
         self.assertEqual(first, canonical_message_json(manifest))
 
+    def test_manifest_rejects_case_colliding_target_paths(self):
+        manifest = self._manifest()
+        manifest.files[1].target_canonical_path = "album/ONE.bin"
+        with self.assertRaisesRegex(
+            TransferIntegrityError, "case-colliding target"
+        ):
+            stage_transfer(
+                manifest,
+                source_root=self.source,
+                sync_root=self.sync,
+                store=self.store,
+                artifact_sources={
+                    "metainfo/source.torrent": self.metainfo
+                },
+            )
+
     def _manifest(self, job_id=None):
         manifest = transfer_pb2.TransferManifest(
             manifest_version=1,