Separate game file downloads and verification with progress timing

agent/ec_agent.py

@@ -30,7 +30,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.2.1"
+VERSION = "1.2.2"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 LIBRARY_DOWNLOAD_WORKERS = 4
@@ -587,6 +587,80 @@ class Uninstaller:
                 time.sleep(15)
 
 
+class LibraryTransfer:
+    """Durable phase progress; throughput counts only work in this attempt."""
+
+    def __init__(self, progress, marker, previous=None):
+        self.progress = progress
+        self.marker = marker
+        self.lock = threading.RLock()
+        prior = (previous or {}).get("transfer") or {}
+        self.prior_phase = prior.get("phase")
+        self.prior_phase_elapsed = prior.get("phaseElapsedSeconds", 0)
+        self.started_at = prior.get("startedAt") or time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+        self.elapsed = prior.get("elapsedSeconds", 0)
+        self.phase_elapsed = prior.get("phaseElapsedSeconds", 0) if prior.get("phase") == progress["status"] else 0
+        self.started = self.phase_started = time.monotonic()
+        self.phase = progress["status"]
+        self.work = 0
+        self.finished = False
+        self.failed_bytes_downloaded = None
+        self.persisted_at = self.started
+        self.persist(force=True)
+
+    def snapshot(self):
+        with self.lock:
+            if not self.finished:
+                now = time.monotonic()
+                sample_seconds = max(0, now - self.phase_started)
+                rate = self.work / sample_seconds if sample_seconds >= 1 and self.work else None
+                done = self.progress["bytesDownloaded"] if self.phase == "downloading" else self.progress["bytesVerified"]
+                self.progress["transfer"] = {
+                    "phase": self.phase, "startedAt": self.started_at,
+                    "elapsedSeconds": self.elapsed + max(0, now - self.started),
+                    "phaseElapsedSeconds": self.phase_elapsed + sample_seconds,
+                    "bytesPerSecond": rate,
+                    "etaSeconds": max(0, self.progress["bytesTotal"] - done) / rate if rate else None,
+                }
+            return {**self.progress, "transfer": self.progress["transfer"].copy()}
+
+    def persist(self, force=False):
+        with self.lock:
+            now = time.monotonic()
+            self.snapshot()
+            if force or now - self.persisted_at >= 5:
+                atomic_json(self.marker, self.progress)
+                self.persisted_at = now
+
+    def advance(self, key, count, work):
+        with self.lock:
+            self.progress[key] += count
+            self.work += work
+            self.persist()
+
+    def begin_verification(self):
+        with self.lock:
+            self.progress.update(status="verifying", bytesVerified=0)
+            if self.phase != "verifying":
+                self.phase_elapsed = self.prior_phase_elapsed if self.prior_phase == "verifying" else 0
+            self.phase = "verifying"
+            self.phase_started = time.monotonic()
+            self.work = 0
+            self.persist(force=True)
+
+    def finish(self, status, error=None):
+        with self.lock:
+            self.snapshot()
+            self.finished = True
+            self.progress["status"] = status
+            if status == "failed" and self.failed_bytes_downloaded is not None:
+                self.progress["bytesDownloaded"] = self.failed_bytes_downloaded
+            self.progress["transfer"]["etaSeconds"] = 0 if status == "ready" else None
+            if error:
+                self.progress["error"] = error
+            self.persist(force=True)
+
+
 class Agent:
     def __init__(self, config, docker=None):
         self.config = config
@@ -599,6 +673,7 @@ class Agent:
         self.stopping = threading.Event()
         self.current_operation = None
         self.library_progress = {}
+        self.library_transfers = {}
         self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
         self.executing_id = None
 
@@ -789,6 +864,8 @@ class Agent:
         except (AgentError, ValueError):
             pass
         libraries, caches = [], []
+        progress = {key: value.copy() for key, value in list(self.library_progress.items())}
+        progress.update({key: transfer.snapshot() for key, transfer in list(self.library_transfers.items())})
         for marker in (self.root / "libraries").glob("*/*/manifest.json"):
             # A crash between writing a manifest and its atomic promotion must
             # never expose the staging directory as a ready library.
@@ -796,12 +873,22 @@ class Agent:
                 continue
             try:
                 value = json.loads(marker.read_text())
+                # Rechecking an existing cache must revoke readiness until its
+                # entire contents have passed verification again.
+                if value["libraryId"] not in progress and (marker.parent / "transfer.json").exists():
+                    saved_progress = json.loads((marker.parent / "transfer.json").read_text())
+                    if saved_progress.get("status") != "ready":
+                        progress[value["libraryId"]] = {**saved_progress, "status": "failed" if saved_progress.get("status") == "failed" else "missing"}
+                if value["libraryId"] in progress:
+                    continue
                 libraries.append({"id": value["libraryId"], "state": "ready"})
-                caches.append({"libraryVersionId": value["libraryId"], "status": "ready", "bytesDownloaded": value["size"], "bytesTotal": value["size"], "manifestSha256": value.get("manifestSha256")})
+                cache = {"libraryVersionId": value["libraryId"], "status": "ready", "bytesDownloaded": value["size"], "bytesVerified": value["size"], "bytesTotal": value["size"], "manifestSha256": value.get("manifestSha256")}
+                if value.get("transfer"):
+                    cache["transfer"] = value["transfer"]
+                caches.append(cache)
             except (OSError, ValueError, KeyError):
                 continue
         ready = {item["id"] for item in libraries}
-        progress = {key: value.copy() for key, value in list(self.library_progress.items()) if key not in ready}
         for marker in (self.root / "libraries").glob("*/.*.partial/source.json"):
             try:
                 value = json.loads(marker.read_text())
@@ -826,7 +913,15 @@ class Agent:
                         retained += min(target.stat().st_size, item["size"])
                 # These are resumable bytes, not verified content. Reporting
                 # them survives restarts without requiring space for two copies.
-                progress[library_id] = {"libraryVersionId": library_id, "status": "missing", "bytesDownloaded": retained, "bytesTotal": size, "manifestSha256": value["manifestSha256"]}
+                cache = {"libraryVersionId": library_id, "status": "missing", "bytesDownloaded": retained, "bytesVerified": 0, "bytesTotal": size, "manifestSha256": value["manifestSha256"]}
+                transfer_marker = marker.parent / "transfer.json"
+                if transfer_marker.exists():
+                    saved_progress = json.loads(transfer_marker.read_text())
+                    if saved_progress.get("manifestSha256") == value["manifestSha256"]:
+                        cache.update(bytesVerified=saved_progress.get("bytesVerified", 0), transfer=saved_progress.get("transfer"))
+                        if saved_progress.get("status") == "failed":
+                            cache.update(status="failed", error=saved_progress.get("error"))
+                progress[library_id] = cache
             except (AgentError, OSError, ValueError, KeyError, TypeError):
                 continue
         progress = list(progress.values())
@@ -1221,14 +1316,14 @@ class Agent:
             raise AgentError("Library transport must use HTTPS")
         return url
 
-    def artifact_open(self, method, path, data=None, extra_headers=None):
+    def artifact_open(self, method, path, data=None, extra_headers=None, timeout=60):
         headers = {"Authorization": "Bearer " + self.config["credential"], **(extra_headers or {})}
         req = urllib.request.Request(self.artifact_url(path), data=data, method=method, headers=headers)
         # Redirects cannot forward host credentials to another origin.
         class NoRedirect(urllib.request.HTTPRedirectHandler):
             def redirect_request(self, req, fp, code, msg, hdrs, newurl):
                 raise AgentError("Artifact redirects are not allowed")
-        return urllib.request.build_opener(NoRedirect()).open(req, timeout=60)
+        return urllib.request.build_opener(NoRedirect()).open(req, timeout=timeout)
 
     def download_library_file(self, item, target, files_url, update_progress, aborted):
         def check_active():
@@ -1241,21 +1336,11 @@ class Agent:
         if offset > item["size"]:
             target.unlink()
             offset = 0
-        checksum = hashlib.sha256()
-        # Hash the saved prefix once so resumed transfers still verify the whole
-        # file. New bytes are hashed as they arrive, avoiding a second disk read.
-        if offset:
-            with target.open("rb") as stream:
-                for chunk in iter(lambda: stream.read(1024 * 1024), b""):
-                    check_active()
-                    checksum.update(chunk)
+        # Saved prefixes and complete files are intentionally not read here.
+        # The second phase verifies every byte after all transfers have ended.
         if offset == item["size"] and target.exists():
-            if checksum.hexdigest() == item["sha256"]:
-                update_progress(item["path"], offset)
-                return
-            target.unlink()
-            offset = 0
-            checksum = hashlib.sha256()
+            update_progress(item["path"], offset)
+            return
         update_progress(item["path"], offset)
         check_active()
         url = files_url.rstrip("/") + "/" + urllib.parse.quote(item["path"], safe="/")
@@ -1269,7 +1354,6 @@ class Agent:
             elif offset:
                 # A server may ignore Range and send the whole file instead.
                 offset = 0
-                checksum = hashlib.sha256()
                 update_progress(item["path"], 0)
             with target.open("ab" if offset else "wb") as output:
                 received = offset
@@ -1282,13 +1366,48 @@ class Agent:
                     if received > item["size"]:
                         raise AgentError("Library download exceeded its manifest size")
                     output.write(chunk)
-                    checksum.update(chunk)
                     update_progress(item["path"], received)
                 output.flush()
                 os.fsync(output.fileno())
-        if target.stat().st_size != item["size"] or checksum.hexdigest() != item["sha256"]:
+        if target.stat().st_size != item["size"]:
+            raise AgentError("Downloaded library file size failed integrity verification")
+
+    def verify_downloaded_files(self, files, targets, transfer, repair=False):
+        invalid = False
+        retained = transfer.progress["bytesDownloaded"]
+        for item in files:
+            target = targets[item["path"]]
+            checksum = hashlib.sha256()
+            actual_size = target.stat().st_size if target.is_file() else 0
+            valid_size = target.is_file() and actual_size == item["size"]
+            if valid_size:
+                with target.open("rb") as stream:
+                    for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+                        checksum.update(chunk)
+                        transfer.advance("bytesVerified", len(chunk), len(chunk))
+            if not valid_size or checksum.hexdigest() != item["sha256"]:
+                invalid = True
+                if repair and target.exists():
+                    # A corrupt complete file (including a corrupt saved
+                    # prefix) must not be resumed indefinitely on every retry.
+                    target.unlink()
+                    actual_size = 0
+                retained -= item["size"] - min(actual_size, item["size"])
+        if invalid:
+            # Verification describes a completed download phase. Keep that
+            # total stable while other files are checked, then publish retained
+            # bytes atomically with failure (including removed corrupt files).
+            transfer.failed_bytes_downloaded = retained
             raise AgentError("Downloaded library file failed integrity verification")
 
+    def publish_library_progress(self, payload, transfer):
+        if payload.get("statusUrl"):
+            # Phase boundaries are useful even when shorter than the heartbeat
+            # interval. Telemetry failure must never undo a verified download.
+            with contextlib.suppress(AgentError, OSError, ValueError):
+                with self.artifact_open("POST", payload["statusUrl"], json.dumps(transfer.snapshot()).encode(), {"Content-Type": "application/json"}, timeout=5) as response:
+                    response.read()
+
     def library_download(self, payload):
         library_id = identifier(payload.get("libraryVersionId", payload.get("libraryId")))
         with self.artifact_open("GET", payload["manifestUrl"]) as response:
@@ -1305,12 +1424,12 @@ class Agent:
         if payload.get("manifestSha256") and payload["manifestSha256"] != manifest_sha:
             raise AgentError("Library manifest differs from the requested immutable version")
         directory = self.library_dir(manifest["game"], library_id)
-        if directory.exists():
-            saved = self.verify_library(directory)
+        existing = directory.exists()
+        if existing:
+            saved = json.loads((directory / "manifest.json").read_text())
             if saved.get("manifestSha256") != manifest_sha:
                 raise AgentError("Published library changed unexpectedly")
-            return {"libraryId": library_id, "state": "ready", "size": saved["size"]}
-        staging = directory.with_name("." + library_id + ".partial")
+        staging = directory if existing else directory.with_name("." + library_id + ".partial")
         staging.mkdir(parents=True, exist_ok=True)
         source_marker = staging / "source.json"
         if source_marker.exists() and json.loads(source_marker.read_text()).get("manifestSha256") != manifest_sha:
@@ -1318,28 +1437,39 @@ class Agent:
         size = sum(item["size"] for item in files)
         atomic_json(source_marker, {"libraryId": library_id, "game": manifest["game"], "size": size, "files": files, "manifestSha256": manifest_sha})
         targets = {}
-        progress = {"libraryVersionId": library_id, "status": "downloading", "bytesDownloaded": 0, "bytesTotal": size, "manifestSha256": manifest_sha}
+        progress = {"libraryVersionId": library_id, "status": "verifying" if existing else "downloading", "bytesDownloaded": size if existing else 0, "bytesVerified": 0, "bytesTotal": size, "manifestSha256": manifest_sha}
         self.library_progress[library_id] = progress
+        marker = staging / "transfer.json"
+        previous = json.loads(marker.read_text()) if marker.exists() else None
+        if previous and (previous.get("manifestSha256") != manifest_sha or previous.get("status") == "ready"):
+            previous = None
+        timing = LibraryTransfer(progress, marker, previous)
+        self.library_transfers[library_id] = timing
         try:
-            if shutil.disk_usage(staging).free < max(0, size - sum(path.stat().st_size for path in staging.rglob("*") if path.is_file())):
-                raise AgentError("Insufficient library storage")
             # Validate every path before starting concurrent writes. In addition
             # to duplicates, a manifest cannot treat a file as a parent folder.
             for item in files:
-                if item["path"] in targets or not isinstance(item["size"], int) or item["size"] < 0 or not re.fullmatch("[0-9a-f]{64}", item["sha256"]):
+                if item["path"] in targets or type(item["size"]) is not int or item["size"] < 0 or not re.fullmatch("[0-9a-f]{64}", item["sha256"]):
                     raise AgentError("Invalid library file manifest")
                 targets[item["path"]] = scoped(staging / "content", item["path"])
             for path in targets:
                 if any(str(parent) in targets for parent in PurePosixPath(path).parents):
                     raise AgentError("Library file paths overlap")
             progress_lock = threading.Lock()
-            offsets = {}
+            offsets = {item["path"]: min(targets[item["path"]].stat().st_size, item["size"]) if targets[item["path"]].exists() else 0 for item in files}
+            if not existing:
+                progress["bytesDownloaded"] = sum(offsets.values())
+            if not existing and shutil.disk_usage(staging).free < size - progress["bytesDownloaded"]:
+                raise AgentError("Insufficient library storage")
+            timing.persist(force=True)
+            self.publish_library_progress(payload, timing)
             aborted = threading.Event()
 
             def update_progress(path, offset):
                 with progress_lock:
-                    progress["bytesDownloaded"] += offset - offsets.get(path, 0)
+                    delta = offset - offsets.get(path, 0)
                     offsets[path] = offset
+                    timing.advance("bytesDownloaded", delta, max(0, delta))
 
             def transfer(item):
                 try:
@@ -1350,21 +1480,44 @@ class Agent:
 
             # Bound sockets, memory and disk pressure while allowing other files
             # to progress during network round trips and per-file fsyncs.
-            with ThreadPoolExecutor(max_workers=min(LIBRARY_DOWNLOAD_WORKERS, len(files))) as pool:
-                futures = [pool.submit(transfer, item) for item in files]
-                for future in as_completed(futures):
-                    future.result()
-            progress["status"] = "verifying"
+            if not existing:
+                # A crash after setting read-only permissions but before the
+                # atomic rename leaves a valid, resumable staging directory.
+                content_root = staging / "content"
+                if content_root.exists():
+                    os.chmod(content_root, 0o755)
+                    for path in content_root.rglob("*"):
+                        if path.is_symlink():
+                            raise AgentError("Symlinks are not allowed in library content")
+                        os.chmod(path, 0o755 if path.is_dir() else 0o644)
+                with ThreadPoolExecutor(max_workers=min(LIBRARY_DOWNLOAD_WORKERS, len(files))) as pool:
+                    futures = [pool.submit(transfer, item) for item in files]
+                    for future in as_completed(futures):
+                        future.result()
+                timing.begin_verification()
+                self.publish_library_progress(payload, timing)
+            self.verify_downloaded_files(files, targets, timing, repair=not existing)
             saved = {"libraryId": library_id, "game": manifest["game"], "size": size, "files": files, "manifestSha256": manifest_sha}
             atomic_json(staging / "manifest.json", saved)
-            for path in (staging / "content").rglob("*"):
-                os.chmod(path, 0o555 if path.is_dir() else 0o444)
-            os.chmod(staging / "content", 0o555)
-            os.replace(staging, directory)
+            if not existing:
+                for path in (staging / "content").rglob("*"):
+                    if path.is_symlink():
+                        raise AgentError("Symlinks are not allowed in library content")
+                    os.chmod(path, 0o555 if path.is_dir() else 0o444)
+                os.chmod(staging / "content", 0o555)
+                os.replace(staging, directory)
+                timing.marker = directory / "transfer.json"
+            timing.finish("ready")
+            saved["transfer"] = timing.snapshot()["transfer"]
+            saved["bytesVerified"] = size
+            atomic_json(directory / "manifest.json", saved)
+            self.publish_library_progress(payload, timing)
             self.library_progress.pop(library_id, None)
+            self.library_transfers.pop(library_id, None)
             return {"libraryId": library_id, "state": "ready", "size": size, "manifestSha256": manifest_sha}
         except Exception:
-            progress.update(status="failed", error="Library download or integrity verification failed")
+            timing.finish("failed", "Library download or integrity verification failed")
+            self.publish_library_progress(payload, timing)
             raise
 
     def library_import(self, payload):
@@ -1561,14 +1714,9 @@ class Agent:
                 finally:
                     self.executing_id = None
             if kind == "library.download":
-                library_id = payload.get("libraryVersionId", payload.get("libraryId"))
-                for game in ("t5", "t6"):
-                    directory = self.library_dir(game, library_id)
-                    if directory.exists():
-                        saved = self.verify_library(directory)
-                        return self.journal.update(operation["id"], "completed", {"libraryId": library_id, "state": "ready", "size": saved["size"], "manifestSha256": saved.get("manifestSha256"), "reconciled": True})
                 # This is a resumable immutable transfer, not a repeated
-                # container command: verified completed files remain untouched.
+                # container command. Full files skip transfer; verification
+                # always starts again so interrupted reads cannot prove ready.
                 self.journal.update(operation["id"], "running")
                 try:
                     return self.journal.update(operation["id"], "completed", self.library_download(payload))

agent/tests/test_agent.py

@@ -277,13 +277,60 @@ class AgentTests(unittest.TestCase):
         for path, content in contents.items():
             self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content" / path).read_bytes(), content)
 
-    def test_fresh_download_hashes_stream_without_rereading_whole_file(self):
-        content = b"game-file" * 200000
-        manifest = self.manifest(content)
-        self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else content)
-        with patch.object(module, "file_digest", side_effect=AssertionError("download reread the file")):
+    def test_all_downloads_finish_before_any_file_is_hashed_and_verified_from_disk(self):
+        contents = {"zone/first.ff": b"game-file" * 200000, "zone/second.ff": b"second-file"}
+        manifest = self.manifest_files(contents)
+        self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else contents[path.removeprefix("/files/")])
+        original_hash, original_open = module.hashlib.sha256, Path.open
+        reads = []
+
+        def hashing(*args, **kwargs):
+            progress = self.agent.library_progress.get("downloaded")
+            if progress:
+                self.assertEqual(progress["status"], "verifying")
+                self.assertEqual(progress["bytesDownloaded"], manifest["sizeBytes"])
+            return original_hash(*args, **kwargs)
+
+        def opening(path, mode="r", *args, **kwargs):
+            if mode == "rb" and path.name.endswith(".ff"):
+                self.assertFalse(self.agent.library_dir("t6", "downloaded").exists())
+                reads.append(path.name)
+            return original_open(path, mode, *args, **kwargs)
+
+        with patch.object(module.hashlib, "sha256", side_effect=hashing), patch.object(Path, "open", opening):
             result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
         self.assertEqual(result["state"], "ready")
+        self.assertCountEqual(reads, ["first.ff", "second.ff"])
+        cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded")
+        self.assertEqual(cache["bytesVerified"], manifest["sizeBytes"])
+        self.assertEqual(cache["transfer"]["phase"], "verifying")
+
+    def test_phase_boundaries_are_reported_even_between_heartbeats_and_status_errors_are_nonfatal(self):
+        content = b"game-file"
+        manifest = self.manifest(content)
+        reports = []
+
+        def fetch(method, path, data=None, extra_headers=None, timeout=60):
+            if path == "/status":
+                self.assertEqual(timeout, 5)
+                reports.append(json.loads(data))
+                raise AgentError("Controller temporarily unavailable")
+            self.assertEqual(timeout, 60)
+            return Response(json.dumps(manifest).encode() if path == "/manifest" else content)
+
+        self.agent.artifact_open = fetch
+        result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files", "statusUrl": "/status"})
+        self.assertEqual(result["state"], "ready")
+        self.assertEqual([item["status"] for item in reports], ["downloading", "verifying", "ready"])
+        self.assertEqual([item["bytesVerified"] for item in reports], [0, 0, len(content)])
+        self.assertEqual(reports[1]["bytesDownloaded"], len(content))
+
+    def test_status_socket_timeout_is_short_without_reducing_download_timeout(self):
+        with patch.object(module.urllib.request, "build_opener") as build_opener:
+            self.agent.artifact_open("GET", "/file")
+            self.assertEqual(build_opener.return_value.open.call_args.kwargs["timeout"], 60)
+            self.agent.artifact_open("POST", "/status", b"{}", timeout=5)
+            self.assertEqual(build_opener.return_value.open.call_args.kwargs["timeout"], 5)
 
     def test_ignored_range_restarts_file_and_checksum(self):
         content = b"complete-game-file"
@@ -313,6 +360,80 @@ class AgentTests(unittest.TestCase):
         with self.assertRaisesRegex(AgentError, "integrity"):
             self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
         self.assertFalse(self.agent.library_dir("t6", "downloaded").exists())
+        self.assertFalse((staging / "test.ff").exists())
+        self.assertEqual(self.agent.library_progress["downloaded"]["bytesDownloaded"], 0)
+        requests = []
+
+        def fetch(method, path, data=None, extra_headers=None):
+            if path != "/manifest":
+                requests.append(extra_headers)
+            return Response(json.dumps(manifest).encode() if path == "/manifest" else content)
+
+        self.agent.artifact_open = fetch
+        self.assertEqual(self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})["state"], "ready")
+        self.assertEqual(requests, [{}])
+
+    def test_all_corrupt_complete_files_are_removed_so_retry_downloads_them_again(self):
+        contents = {"zone/first.ff": b"first", "zone/second.ff": b"second", "zone/good.ff": b"good"}
+        manifest = self.manifest_files(contents)
+        staging = self.agent.root / "libraries/t6/.downloaded.partial/content/zone"
+        staging.mkdir(parents=True)
+        for name, content in contents.items():
+            (staging / Path(name).name).write_bytes(content if name.endswith("good.ff") else b"x" * len(content))
+        requests = []
+
+        def fetch(method, path, data=None, extra_headers=None):
+            requests.append(path)
+            return Response(json.dumps(manifest).encode() if path == "/manifest" else contents[path.removeprefix("/files/")])
+
+        self.agent.artifact_open = fetch
+        payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}
+        with self.assertRaisesRegex(AgentError, "integrity"):
+            self.agent.library_download(payload)
+        self.assertEqual(requests, ["/manifest"])
+        self.assertEqual([path.name for path in staging.iterdir()], ["good.ff"])
+        self.assertEqual(self.agent.library_download(payload)["state"], "ready")
+        self.assertCountEqual(requests, ["/manifest", "/manifest", "/files/zone/first.ff", "/files/zone/second.ff"])
+
+    def test_verification_keeps_download_phase_complete_until_failure_reports_retained_bytes(self):
+        contents = {"zone/first.ff": b"first", "zone/second.ff": b"second"}
+        for mode in ("staged-corrupt", "existing-missing", "existing-short"):
+            with self.subTest(mode=mode):
+                manifest = {**self.manifest_files(contents), "id": mode}
+                directory = self.agent.library_dir("t6", mode)
+                if mode.startswith("existing"):
+                    (directory / "content/zone").mkdir(parents=True)
+                    (directory / "content/zone/second.ff").write_bytes(contents["zone/second.ff"])
+                    if mode == "existing-short":
+                        (directory / "content/zone/first.ff").write_bytes(b"f")
+                    module.atomic_json(directory / "manifest.json", {"libraryId": mode, "game": "t6", "size": manifest["sizeBytes"], "files": manifest["files"], "manifestSha256": manifest["manifestSha256"]})
+
+                def fetch(method, path, **kwargs):
+                    if path == "/manifest":
+                        return Response(json.dumps(manifest).encode())
+                    content = contents[path.removeprefix("/files/")]
+                    return Response(b"x" * len(content) if path.endswith("first.ff") else content)
+
+                self.agent.artifact_open = fetch
+                original_open = Path.open
+                snapshots = []
+
+                def opening(path, open_mode="r", *args, **kwargs):
+                    open_mode = kwargs.pop("mode", open_mode)
+                    if open_mode == "rb" and path.name == "second.ff":
+                        cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == mode)
+                        snapshots.append(cache)
+                        self.assertEqual(cache["status"], "verifying")
+                        self.assertEqual(cache["bytesDownloaded"], cache["bytesTotal"])
+                        self.assertEqual(cache["transfer"]["phase"], "verifying")
+                    return original_open(path, open_mode, *args, **kwargs)
+
+                with patch.object(Path, "open", opening), self.assertRaisesRegex(AgentError, "integrity"):
+                    self.agent.library_download({"libraryVersionId": mode, "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
+                self.assertEqual(len(snapshots), 1)
+                failed = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == mode)
+                self.assertEqual(failed["status"], "failed")
+                self.assertEqual(failed["bytesDownloaded"], len(contents["zone/second.ff"]) + (1 if mode == "existing-short" else 0))
 
     def test_parallel_failure_retains_partial_files_for_verified_resume(self):
         contents = {"zone/first.ff": b"first-game-file", "zone/second.ff": b"second-game-file"}
@@ -369,7 +490,13 @@ class AgentTests(unittest.TestCase):
         self.agent.journal.close()
         self.agent = Agent(self.config, self.docker)
         cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded")
-        self.assertEqual(cache, {"libraryVersionId": "downloaded", "status": "missing", "bytesDownloaded": 8, "bytesTotal": len(content), "manifestSha256": manifest["manifestSha256"]})
+        self.assertEqual(cache["status"], "failed")
+        self.assertEqual(cache["bytesDownloaded"], 8)
+        self.assertEqual(cache["bytesVerified"], 0)
+        self.assertEqual(cache["bytesTotal"], len(content))
+        self.assertEqual(cache["manifestSha256"], manifest["manifestSha256"])
+        self.assertEqual(cache["transfer"]["phase"], "downloading")
+        self.assertIsNone(cache["transfer"]["etaSeconds"])
 
         def fetch(method, path, data=None, extra_headers=None):
             if path == "/manifest":
@@ -405,6 +532,139 @@ class AgentTests(unittest.TestCase):
         self.assertEqual(caches[0]["status"], "downloading")
         self.assertEqual(caches[0]["bytesDownloaded"], 3)
 
+    def test_interrupted_verification_restarts_from_disk_without_redownloading_and_keeps_elapsed(self):
+        content = b"complete-game-file"
+        manifest = self.manifest(content)
+        payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}
+        self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else content)
+
+        def interrupted(files, targets, transfer, repair=False):
+            transfer.advance("bytesVerified", 5, 5)
+            transfer.persist(force=True)
+            raise KeyboardInterrupt("Simulated process termination")
+
+        with patch.object(self.agent, "verify_downloaded_files", side_effect=interrupted), self.assertRaises(KeyboardInterrupt):
+            self.agent.library_download(payload)
+        previous = self.agent.library_transfers["downloaded"].snapshot()
+        self.assertFalse(self.agent.library_dir("t6", "downloaded").exists())
+        self.agent.journal.close()
+        self.agent = Agent(self.config, self.docker)
+        interrupted_cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded")
+        self.assertEqual(interrupted_cache["status"], "missing")
+        self.assertEqual(interrupted_cache["bytesDownloaded"], len(content))
+        self.assertEqual(interrupted_cache["bytesVerified"], 5)
+        requests = []
+
+        def fetch(method, path, **kwargs):
+            requests.append(path)
+            return Response(json.dumps(manifest).encode())
+
+        original_verify = self.agent.verify_downloaded_files
+
+        def verifying(files, targets, transfer, repair=False):
+            self.assertEqual(transfer.snapshot()["bytesVerified"], 0)
+            self.assertIsNone(transfer.snapshot()["transfer"]["etaSeconds"])
+            return original_verify(files, targets, transfer, repair)
+
+        self.agent.artifact_open = fetch
+        with patch.object(self.agent, "verify_downloaded_files", side_effect=verifying):
+            self.assertEqual(self.agent.library_download(payload)["state"], "ready")
+        self.assertEqual(requests, ["/manifest"])
+        cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded")
+        self.assertEqual(cache["bytesVerified"], len(content))
+        self.assertEqual(cache["transfer"]["startedAt"], previous["transfer"]["startedAt"])
+        self.assertGreaterEqual(cache["transfer"]["elapsedSeconds"], interrupted_cache["transfer"]["elapsedSeconds"])
+        self.assertGreaterEqual(cache["transfer"]["phaseElapsedSeconds"], interrupted_cache["transfer"]["phaseElapsedSeconds"])
+        self.agent.journal.close()
+        self.agent = Agent(self.config, self.docker)
+        self.assertEqual(next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded"), cache)
+
+    def test_crash_before_atomic_promotion_reverifies_read_only_staging_without_network_transfers(self):
+        content = b"complete-game-file"
+        manifest = self.manifest(content)
+        payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}
+        self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else content)
+        original_replace = module.os.replace
+
+        def replace(source, destination):
+            if Path(destination) == self.agent.library_dir("t6", "downloaded"):
+                raise KeyboardInterrupt("Simulated process termination before promotion")
+            return original_replace(source, destination)
+
+        with patch.object(module.os, "replace", side_effect=replace), self.assertRaises(KeyboardInterrupt):
+            self.agent.library_download(payload)
+        staging = self.agent.root / "libraries/t6/.downloaded.partial/content"
+        self.assertEqual(staging.stat().st_mode & 0o777, 0o555)
+        self.agent.journal.close()
+        self.agent = Agent(self.config, self.docker)
+        requests = []
+
+        def fetch(method, path, **kwargs):
+            requests.append(path)
+            return Response(json.dumps(manifest).encode())
+
+        self.agent.artifact_open = fetch
+        self.assertEqual(self.agent.library_download(payload)["state"], "ready")
+        self.assertEqual(requests, ["/manifest"])
+        self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content/zone/test.ff").read_bytes(), content)
+
+    def test_existing_cache_reverification_is_visible_and_failure_revokes_readiness_after_restart(self):
+        saved = json.loads((self.library / "manifest.json").read_text())
+        manifest = {**saved, "id": self.library_id}
+        self.agent.artifact_open = lambda *args, **kwargs: Response(json.dumps(manifest).encode())
+        original_verify = self.agent.verify_downloaded_files
+
+        def verifying(files, targets, transfer, repair=False):
+            inventory = self.agent.inventory()
+            cache = next(item for item in inventory["libraryCaches"] if item["libraryVersionId"] == self.library_id)
+            self.assertEqual(cache["status"], "verifying")
+            self.assertEqual(cache["bytesVerified"], 0)
+            self.assertNotIn({"id": self.library_id, "state": "ready"}, inventory["libraries"])
+            return original_verify(files, targets, transfer, repair)
+
+        (self.library / "content/game.ff").write_bytes(b"corrupt!!")
+        with patch.object(self.agent, "verify_downloaded_files", side_effect=verifying), self.assertRaisesRegex(AgentError, "integrity"):
+            self.agent.library_download({"libraryVersionId": self.library_id, "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
+        self.agent.journal.close()
+        self.agent = Agent(self.config, self.docker)
+        inventory = self.agent.inventory()
+        self.assertNotIn({"id": self.library_id, "state": "ready"}, inventory["libraries"])
+        self.assertEqual(next(item for item in inventory["libraryCaches"] if item["libraryVersionId"] == self.library_id)["status"], "failed")
+
+    def test_elapsed_and_phase_eta_use_new_work_only_and_freeze_at_completion_or_failure(self):
+        for terminal_status in ("ready", "failed"):
+            with self.subTest(terminal_status=terminal_status):
+                now = [100.0]
+                progress = {"libraryVersionId": "downloaded", "status": "downloading", "bytesDownloaded": 80, "bytesVerified": 0, "bytesTotal": 100}
+                previous = {"transfer": {"phase": "downloading", "startedAt": "2026-10-04T00:00:00Z", "elapsedSeconds": 20, "phaseElapsedSeconds": 20}}
+                with patch.object(module.time, "monotonic", side_effect=lambda: now[0]):
+                    transfer = module.LibraryTransfer(progress, self.agent.root / "agent/timing.json", previous)
+                    self.assertIsNone(transfer.snapshot()["transfer"]["etaSeconds"])
+                    now[0] += 0.5
+                    transfer.advance("bytesDownloaded", 5, 5)
+                    self.assertIsNone(transfer.snapshot()["transfer"]["etaSeconds"])
+                    now[0] += 1.5
+                    snapshot = transfer.snapshot()["transfer"]
+                    self.assertEqual(snapshot["bytesPerSecond"], 2.5)
+                    self.assertEqual(snapshot["etaSeconds"], 6)
+                    self.assertEqual(snapshot["elapsedSeconds"], 22)
+                    now[0] += 1
+                    transfer.advance("bytesDownloaded", 15, 15)
+                    transfer.begin_verification()
+                    self.assertEqual(transfer.snapshot()["transfer"]["phaseElapsedSeconds"], 0)
+                    self.assertIsNone(transfer.snapshot()["transfer"]["etaSeconds"])
+                    now[0] += 2
+                    transfer.advance("bytesVerified", 40, 40)
+                    snapshot = transfer.snapshot()["transfer"]
+                    self.assertEqual(snapshot["bytesPerSecond"], 20)
+                    self.assertEqual(snapshot["etaSeconds"], 3)
+                    self.assertEqual(snapshot["elapsedSeconds"], 25)
+                    transfer.finish(terminal_status)
+                    final = transfer.snapshot()
+                    now[0] += 100
+                    self.assertEqual(transfer.snapshot(), final)
+                    self.assertEqual(final["transfer"]["etaSeconds"], 0 if terminal_status == "ready" else None)
+
     def test_partial_inventory_rejects_changed_manifest_and_symlinked_content(self):
         content = b"complete-game-file"
         manifest = self.manifest(content)