修复 CI 缓存对象误判与产物下载完整性检查

按 ZIP 成员内容和元数据识别等价缓存对象，保留真实冲突拒绝规则。
核对产物大小与 ZIP 完整性，为截断和临时传输失败增加有界重试。
补充回归测试并同步部署文档与共享记忆。
通过 80 项缓存测试、17 项工作流测试及真实产物和 HTTP 截断验证。

deploy/container/README.md

@@ -112,6 +112,10 @@ runner 配置保留原 `ubuntu-latest` 映射，`genarrative-ci` 继续映射到
 
 ### Rust 测试组编译对象快照
 
+宿主下载每份 artifact 时核对 Gitea `size_in_bytes`、响应 `Content-Length`（若提供）和实际字节数，并在下载完成后立即检查 ZIP 格式与 CRC。截断、损坏归档或临时传输故障最多尝试 3 次，每次重新获取签名下载地址；不向签名地址转发 API Token。重试仍失败则删除临时下载并保留现役镜像，不能把 EOF 当作完整下载成功。
+
+同一 sccache key 的完整对象 SHA 不同时，不直接视为编译结果不同：sccache 对象 ZIP 的成员写入顺序可能不同。仅对不同 job 新增对象之间的冲突，按成员名核对内容 SHA-256、权限及 ZIP 元数据，全部一致才保留一份原始对象并更新使用时间；真实内容差异、异常 ZIP 和继承对象冲突仍拒绝。此比较不改写缓存 key 或对象，不依赖 CRC 代替内容校验。
+
 自动维护由宿主 systemd timer 调用 `scripts/maintain-gitea-rust-cache.py`，只管理 Gitea CI 测试镜像，不修改 Jenkins、生产发布、本地开发或客户端发行构建。六个 Rust job 仅在 master push 中导出本次 CI 新增的 sccache 对象；已命中的继承对象只上传新近使用时间，通过 Gitea 原生 V4 artifact 接口上传；PR 不发布。维护器选择已结束且六组产物完整的最新 master run，校验提交、任务尝试、工具链与来源镜像，与六组实际使用的同一镜像快照合并去重，并按新近使用时间限制快照总容量为 4 GiB，然后从无对象缓存基础镜像组装新镜像，**不重复执行 Cargo 预热编译，也不要求源 run 事先全绿**。缺组、取消或校验失败时保留现役版，不混合不同 run 的对象来假装完整快照。
 
 维护 journal 分阶段记录来源 run、基础镜像重建或复用、artifact 下载、对象合并、镜像组装校验、导出、载入及空闲等待；长操作记录开始和结束耗时，失败输出对应私有 `artifacts/<source-sha>/build.log` 路径。构建的详细下载与 Docker 输出仍只写该日志，不回显 Token、命令环境或认证配置。

docs/project-memory/shared-memory/development-workflow.md

@@ -96,6 +96,8 @@ SpacetimeDB 任务统一先读取 `.codex/skills/genarrative-spacetimedb/SKILL.m
 
 ## Gitea CI 依赖闭合
 
+缓存维护下载必须核对 artifact 元数据大小与实际响应，并立即检查 ZIP 完整性；有长度上限不等于能检测短读。网络或归档损坏最多重试 3 次且重新取签名链接。sccache 同 key 的不同 ZIP 成员排列会改变整个对象 SHA；新增对象去重仅在成员内容 SHA、权限和 ZIP 元数据均一致时接受排列差异，真实内容及继承对象冲突仍拒绝。禁止通过任取一份冲突对象绕过完整性契约。
+
 Buildx 0.30.1 的 `inspect` 不支持 `--format`，builder 驱动校验读取普通输出的 `Driver:` 字段。相关命令须在宿主真实插件上验证；测试替身应拒绝不支持的参数，避免把模拟命令成功误当兼容性证据。
 
 Gitea 基础镜像通过专用 `genarrative-ci-images` Buildx builder 持久复用 Cargo/npm 下载缓存；稳定 cache mount 与 commit、lock 哈希无关，以 `sharing=locked` 隔离并发写入，仅供可信宿主构建、不开放给 PR。最终镜像显式物化当前依赖下载快照，仍不包含 node_modules/target 或上一版 sccache 层。首次可用 `seed-downloads` 从可信完整 Image ID 提取包缓存，操作账号须与维护服务一致；部署要求及 builder GC 空间目标见 `deploy/container/README.md`。构建上下文必须覆盖 AGC vendor 与编辑器 bridge 的全部本地 path manifest，普通源码变化不应使依赖层失效。维护 journal 提供阶段耗时和失败 build.log 定位。

scripts/gitea_cache_snapshot.py

@@ -167,6 +167,14 @@ def _snapshot_member(archive: Path) -> tuple[zipfile.ZipFile, zipfile.ZipInfo]:
         raise
 
 
+def validate_artifact_zip(archive: Path) -> None:
+    """Validate the bounded stored ZIP envelope before consuming an artifact."""
+    bundle, _ = _snapshot_member(archive)
+    with bundle:
+        if bundle.testzip() is not None:
+            raise SnapshotError("artifact ZIP CRC check failed")
+
+
 def _tar_stream(archive: Path):
     bundle, member = _snapshot_member(archive)
     try:
@@ -440,6 +448,94 @@ def _file_sha256(path: Path) -> str:
             digest.update(chunk)
 
 
+def _object_zip_signature(stream: BinaryIO, expected: _Object) -> tuple:
+    """只允许 sccache 对象的 ZIP 成员排列不同，不放宽内容或元数据校验。"""
+    with tempfile.TemporaryFile() as temporary:
+        digest = hashlib.sha256()
+        total = 0
+        while chunk := stream.read(1024 * 1024):
+            total += len(chunk)
+            if total > expected.size:
+                raise SnapshotError("cache object grew while comparing ZIP contents")
+            digest.update(chunk)
+            temporary.write(chunk)
+        if (total, digest.hexdigest()) != (expected.size, expected.sha256):
+            raise SnapshotError("cache object checksum changed while comparing ZIP contents")
+        temporary.seek(0)
+        with zipfile.ZipFile(temporary) as bundle:
+            infos = bundle.infolist()
+            if (not infos or len(infos) > MAX_OBJECTS
+                    or len({info.filename for info in infos}) != len(infos)
+                    or sum(info.file_size for info in infos) > expected.size):
+                raise SnapshotError("invalid sccache ZIP member set")
+            members = []
+            for info in sorted(infos, key=lambda entry: entry.filename):
+                _safe_zip_path(info.filename)
+                mode = info.external_attr >> 16
+                if (info.is_dir() or info.flag_bits & 1 or stat.S_ISLNK(mode)
+                        or stat.S_IFMT(mode) not in (0, stat.S_IFREG)
+                        or info.compress_type != zipfile.ZIP_STORED):
+                    raise SnapshotError("unsupported sccache ZIP member")
+                with bundle.open(info) as contents:
+                    size, checksum = _hash_stream(contents, info.file_size)
+                members.append((
+                    info.filename, size, checksum, info.CRC, info.compress_size,
+                    info.compress_type, info.date_time, info.flag_bits,
+                    info.external_attr, info.internal_attr, info.create_system,
+                    info.create_version, info.extract_version, info.reserved,
+                    info.extra, info.comment,
+                ))
+            return bundle.comment, tuple(members)
+
+
+def _equivalent_delta_paths(archives: Sequence[_Archive]) -> set[str]:
+    variants: dict[str, set[tuple[int, str | None]]] = {}
+    for archive in archives:
+        for obj in archive.objects:
+            variants.setdefault(obj.path, set()).add((obj.size, obj.sha256))
+    conflicts = {path for path, versions in variants.items() if len(versions) > 1}
+    if not conflicts:
+        return conflicts
+    for path in conflicts:
+        if len({size for size, _ in variants[path]}) != 1:
+            raise SnapshotError(f"conflicting content for duplicate object: {path}")
+    signatures = {}
+    # 每份归档最多额外顺序读取一次，仅将冲突对象暂存到磁盘供 ZIP 随机读取。
+    for archive in archives:
+        wanted = {obj.path: obj for obj in archive.objects if obj.path in conflicts}
+        if not wanted:
+            continue
+        if _archive_signature(archive.input.archive) != archive.signature:
+            raise SnapshotError("artifact archive changed while comparing cache objects")
+        bundle, raw, tar = _tar_stream(archive.input.archive)
+        try:
+            for member in tar:
+                expected = wanted.get(member.name)
+                if expected is None:
+                    continue
+                if not member.isreg() or member.size != expected.size:
+                    raise SnapshotError("cache object changed while comparing ZIP contents")
+                stream = tar.extractfile(member)
+                if stream is None:
+                    raise SnapshotError("cache object is missing while comparing ZIP contents")
+                try:
+                    with stream:
+                        signature = _object_zip_signature(stream, expected)
+                except (SnapshotError, zipfile.BadZipFile, NotImplementedError) as error:
+                    raise SnapshotError(f"conflicting content for duplicate object: {member.name}") from error
+                if member.name in signatures and signatures[member.name] != signature:
+                    raise SnapshotError(f"conflicting content for duplicate object: {member.name}")
+                signatures[member.name] = signature
+                del wanted[member.name]
+            if wanted:
+                raise SnapshotError("cache objects disappeared while comparing ZIP contents")
+        finally:
+            tar.close()
+            raw.close()
+            bundle.close()
+    return conflicts
+
+
 def _select_objects(
     archives: Sequence[_Archive],
     base_root: Path,
@@ -447,6 +543,7 @@ def _select_objects(
     maximum: int,
 ) -> tuple[_Object, ...]:
     merged = dict(base)
+    equivalent_paths = _equivalent_delta_paths(archives)
     delta_paths = {obj.path for archive in archives for obj in archive.objects}
     touched_paths = {touch.path for archive in archives for touch in archive.touched}
     overlap = delta_paths & touched_paths
@@ -490,7 +587,8 @@ def _select_objects(
                     mtime_ns=max(current.mtime_ns, candidate.mtime_ns),
                     source_index=None,
                 )
-            elif (current.size, current.sha256) != (candidate.size, candidate.sha256):
+            elif ((current.size, current.sha256) != (candidate.size, candidate.sha256)
+                  and candidate.path not in equivalent_paths):
                 raise SnapshotError(f"conflicting content for duplicate object: {candidate.path}")
             elif candidate.mtime_ns > current.mtime_ns:
                 merged[candidate.path] = _Object(

scripts/maintain-gitea-rust-cache.py

@@ -5,6 +5,7 @@ import argparse
 import contextlib
 import datetime
 import hashlib
+import http.client
 import json
 import os
 from pathlib import Path
@@ -18,8 +19,11 @@ import time
 import urllib.error
 import urllib.parse
 import urllib.request
+import zipfile
 
-from gitea_cache_snapshot import ArtifactIdentity, ArtifactInput, merge_snapshots
+from gitea_cache_snapshot import (
+    ArtifactIdentity, ArtifactInput, SnapshotError, merge_snapshots, validate_artifact_zip,
+)
 from gitea_cache_upload_cleanup import cleanup_upload_chunks
 
 
@@ -46,6 +50,7 @@ RUST_JOBS = set(RUST_JOB_IDS)
 ARTIFACT_PREFIX = "rust-cache-v1-"
 MAX_DOWNLOAD = 4 * 1024 ** 3 + 129 * 1024 ** 2
 EXPORT_STEP = "Publish master Rust cache artifact"
+DOWNLOAD_ATTEMPTS = 3
 
 
 def now():
@@ -183,35 +188,90 @@ class Api:
             raise RuntimeError(f"Gitea API HTTP {error.code}") from None
         return content if raw else (json.loads(content) if content else None)
 
-    def download(self, path, destination):
+    def download(self, path, destination, expected_size):
-        """REST V4 archive redirects to a signed URL; never forward the API token."""
+        """Fetch one artifact archive, retrying incomplete signed-URL transfers."""
+        if (isinstance(expected_size, bool) or not isinstance(expected_size, int)
+                or expected_size <= 0):
+            raise RuntimeError("artifact has invalid size metadata")
+        if expected_size > MAX_DOWNLOAD:
+            raise RuntimeError("artifact exceeds per-job size limit")
+
+        for attempt in range(DOWNLOAD_ATTEMPTS):
+            try:
+                self._download_once(path, destination, expected_size)
+                return
+            except RetryableArtifactDownload as error:
+                destination.unlink(missing_ok=True)
+                log(f"artifact download attempt {attempt + 1}/{DOWNLOAD_ATTEMPTS} failed: {error}")
+                if attempt + 1 == DOWNLOAD_ATTEMPTS:
+                    raise RuntimeError("artifact download remained incomplete after retries") from error
+                time.sleep(1 << attempt)
+            except Exception:
+                destination.unlink(missing_ok=True)
+                raise
+
+    def _download_once(self, path, destination, expected_size):
+        """Obtain a fresh signed URL and validate its complete ZIP response."""
         token = self.token_file.read_text().strip()
         url = self.url + "/" + path.lstrip("/")
         opener = urllib.request.build_opener(NoRedirect())
         request = urllib.request.Request(url, headers={"Authorization": "token " + token})
+        target = None
+        try:
             try:
                 response = opener.open(request, timeout=60)
             except urllib.error.HTTPError as error:
                 error.close()
                 if error.code not in (301, 302, 303, 307, 308):
+                    if error.code >= 500:
+                        raise RetryableArtifactDownload(f"artifact download HTTP {error.code}") from None
                     raise RuntimeError(f"artifact download HTTP {error.code}") from None
                 target = urllib.parse.urljoin(url, error.headers.get("Location", ""))
                 parsed, origin = urllib.parse.urlsplit(target), urllib.parse.urlsplit(self.url)
                 if (parsed.scheme != "https" or parsed.netloc != origin.netloc
                         or parsed.username or parsed.password or target == url):
                     raise RuntimeError("artifact redirect must stay on configured HTTPS Gitea origin") from None
-            response = opener.open(target, timeout=60)
+            if target is not None:
                 try:
-            with response, destination.open("wb") as out:
+                    response = opener.open(target, timeout=60)
+                except urllib.error.HTTPError as error:
+                    error.close()
+                    if error.code >= 500:
+                        raise RetryableArtifactDownload(f"artifact download HTTP {error.code}") from None
+                    raise RuntimeError(f"artifact download HTTP {error.code}") from None
+            with response:
+                content_length = getattr(response, "headers", {}).get("Content-Length")
+                if content_length is not None:
+                    try:
+                        content_length = int(content_length)
+                    except (TypeError, ValueError):
+                        raise RetryableArtifactDownload("artifact response has invalid Content-Length") from None
+                    if content_length > MAX_DOWNLOAD:
+                        raise RuntimeError("artifact exceeds per-job size limit")
+                    if content_length != expected_size:
+                        raise RetryableArtifactDownload(
+                            f"artifact response size differs: expected {expected_size} bytes, "
+                            f"Content-Length is {content_length}")
+                with destination.open("wb") as out:
                     total = 0
                     while chunk := response.read(1024 * 1024):
                         total += len(chunk)
                         if total > MAX_DOWNLOAD:
                             raise RuntimeError("artifact exceeds per-job size limit")
+                        if total > expected_size:
+                            raise RetryableArtifactDownload(
+                                f"artifact response exceeds metadata: expected {expected_size} bytes, "
+                                f"received at least {total}")
                         out.write(chunk)
-        except Exception:
+            if total != expected_size:
-            destination.unlink(missing_ok=True)
+                raise RetryableArtifactDownload(
-            raise
+                    f"artifact response is truncated: expected {expected_size} bytes, received {total}")
+            try:
+                validate_artifact_zip(destination)
+            except (SnapshotError, zipfile.BadZipFile):
+                raise RetryableArtifactDownload("artifact response is not a valid ZIP") from None
+        except (http.client.HTTPException, urllib.error.URLError, OSError) as error:
+            raise RetryableArtifactDownload("artifact transfer failed") from error
 
     def pages(self, path, key):
         separator = "&" if "?" in path else "?"
@@ -230,6 +290,10 @@ class NoRedirect(urllib.request.HTTPRedirectHandler):
         return None
 
 
+class RetryableArtifactDownload(RuntimeError):
+    """A signed archive transfer may be retried with a newly issued URL."""
+
+
 class Maintenance:
     def __init__(self, config):
         self.config = config
@@ -371,7 +435,10 @@ class Maintenance:
                 if len(used) != 1:
                     break
                 images.update(used)
-                selected.append({"id": matches[0]["id"], "name": name,
+                size = matches[0].get("size_in_bytes")
+                if isinstance(size, bool) or not isinstance(size, int) or size <= 0 or size > MAX_DOWNLOAD:
+                    break
+                selected.append({"id": matches[0]["id"], "name": name, "size_in_bytes": size,
                                  "job": RUST_JOB_IDS[job["name"]], "attempt": job["run_attempt"]})
             if len(selected) != len(RUST_JOB_IDS) or len(images) != 1:
                 continue
@@ -431,7 +498,8 @@ class Maintenance:
                 archive = work / (str(export["id"]) + ".zip")
                 with operation(f"download cache artifact job={export['job']} attempt={export['attempt']}",
                                build_log=build_log):
-                    self.api.download(self.repo_api + f'/actions/artifacts/{export["id"]}/zip', archive)
+                    self.api.download(self.repo_api + f'/actions/artifacts/{export["id"]}/zip', archive,
+                                      export["size_in_bytes"])
                 inputs.append(ArtifactInput(archive, ArtifactIdentity(
                     self.config["repository"], source["run_id"], export["attempt"], export["job"], sha)))
             snapshot = work / "snapshot"

scripts/test_gitea_cache_maintenance.py

@@ -11,6 +11,7 @@ import re
 import tempfile
 import unittest
 from unittest.mock import patch
+import zipfile
 
 
 SCRIPT = Path(__file__).with_name("maintain-gitea-rust-cache.py")
@@ -29,6 +30,26 @@ def runner_config(image: str) -> str:
     return f'runners:\n  - "genarrative-ci:docker://{image}"\n'
 
 
+def zip_bytes(*, size=None) -> bytes:
+    """Make a valid stored ZIP, optionally padded to an exact download size."""
+    content = b"x" * 40000 if size else b"cache artifact"
+    output = io.BytesIO()
+    with zipfile.ZipFile(output, "w", zipfile.ZIP_STORED) as archive:
+        archive.writestr("snapshot.tar", content)
+    value = output.getvalue()
+    if size is None:
+        return value
+    if not len(value) < size <= len(value) + 65535:
+        raise AssertionError("requested ZIP size cannot be represented by its comment")
+    output = io.BytesIO(value)
+    with zipfile.ZipFile(output, "a") as archive:
+        archive.comment = b"p" * (size - len(value))
+    value = output.getvalue()
+    if len(value) != size:
+        raise AssertionError("ZIP comment did not produce the requested size")
+    return value
+
+
 class FakeApi:
     def __init__(self):
         self.requests: list[str] = []
@@ -299,7 +320,7 @@ class GiteaCacheMaintenanceTest(unittest.TestCase):
                  "steps": [{"name": maintenance_module.EXPORT_STEP, "conclusion": "success"}]}
                 for i, name in enumerate(maintenance_module.RUST_JOB_IDS, start=1)]
         artifacts = [{"id": job["id"], "name": "rust-cache-v1-" + maintenance_module.RUST_JOB_IDS[job["name"]]
-                      + "-attempt-1", "expired": False, "workflow_run": run} for job in jobs]
+                      + "-attempt-1", "size_in_bytes": 10, "expired": False, "workflow_run": run} for job in jobs]
         class Api:
             def pages(self, path, key):
                 return iter({"workflow_runs": [run], "jobs": jobs, "artifacts": artifacts}[key])
@@ -320,6 +341,7 @@ class GiteaCacheMaintenanceTest(unittest.TestCase):
         result = self.select_source(instance)
         self.assertEqual(result["run_id"], 44)
         self.assertEqual(len(result["exports"]), 6)
+        self.assertEqual({item["size_in_bytes"] for item in result["exports"]}, {10})
         run["event"] = "pull_request"
         self.assertIsNone(self.select_source(instance))
         run["event"] = "push"
@@ -375,7 +397,7 @@ class GiteaCacheMaintenanceTest(unittest.TestCase):
                 return "rustc test\n"
             return ""
         instance.docker = docker
-        instance.api.download = lambda path, destination: destination.write_bytes(b"download")
+        instance.api.download = lambda path, destination, expected_size: destination.write_bytes(b"download")
         def merge(inputs, output, **kwargs):
             self.assertEqual(len(inputs), 6)
             self.assertEqual(kwargs["expected_inherited_source_sha"], "b" * 40)
@@ -419,23 +441,111 @@ class GiteaCacheMaintenanceTest(unittest.TestCase):
     def test_signed_download_drops_token_and_rejects_other_origins(self):
         api = maintenance_module.Api(self.config["api_url"], self.token)
         destination = self.root / "artifact.zip"
+        archive = zip_bytes()
         seen = []
         target = ["https://gitea.example.test/api/v1/repos/team/project/actions/artifacts/1/zip/raw?sig=test"]
+
+        class Response(io.BytesIO):
+            def __init__(self, value):
+                super().__init__(value)
+                self.headers = {"Content-Length": str(len(value))}
+
         class Opener:
             def open(self, request, timeout):
                 seen.append(request)
                 if not isinstance(request, str):
                     raise maintenance_module.urllib.error.HTTPError(request.full_url, 302, "Found", {"Location": target[0]}, None)
-                return io.BytesIO(b"archive")
+                return Response(archive)
+
         with patch.object(maintenance_module.urllib.request, "build_opener", return_value=Opener()):
-            api.download("repos/team/project/actions/artifacts/1/zip", destination)
+            api.download("repos/team/project/actions/artifacts/1/zip", destination, len(archive))
-            self.assertEqual(destination.read_bytes(), b"archive")
+            self.assertEqual(destination.read_bytes(), archive)
             self.assertEqual(seen[0].get_header("Authorization"), "token test-token")
             self.assertIsInstance(seen[1], str)  # signed URL request has no Authorization header
             target[0] = "https://other.example.test/download"
             with self.assertRaisesRegex(RuntimeError, "configured HTTPS Gitea origin"):
-                api.download("repos/team/project/actions/artifacts/1/zip", self.root / "rejected.zip")
+                api.download("repos/team/project/actions/artifacts/1/zip", self.root / "rejected.zip", len(archive))
         self.assertFalse((self.root / "rejected.zip").exists())
+        self.assertEqual(len(seen), 3)  # invalid redirect is deterministic and is not retried
+
+    def test_download_retries_transport_and_truncated_response_with_fresh_signed_urls(self):
+        api = maintenance_module.Api(self.config["api_url"], self.token)
+        destination = self.root / "artifact.zip"
+        archive = zip_bytes(size=100000)
+        seen, signed_responses = [], []
+        origin_requests = 0
+        target = "https://gitea.example.test/api/v1/repos/team/project/actions/artifacts/1/zip/raw?sig=test"
+
+        class Response(io.BytesIO):
+            def __init__(self, value):
+                super().__init__(value)
+                self.headers = {"Content-Length": "100000"}
+
+        class Opener:
+            def open(self, request, timeout):
+                nonlocal origin_requests
+                seen.append(request)
+                if not isinstance(request, str):
+                    origin_requests += 1
+                    if origin_requests == 1:
+                        raise maintenance_module.urllib.error.URLError("connection reset")
+                    raise maintenance_module.urllib.error.HTTPError(request.full_url, 302, "Found", {"Location": target}, None)
+                signed_responses.append(request)
+                return Response(b"x" * 512 if len(signed_responses) == 1 else archive)
+
+        with patch.object(maintenance_module.urllib.request, "build_opener", return_value=Opener()), \
+             patch.object(maintenance_module.time, "sleep") as sleep:
+            api.download("repos/team/project/actions/artifacts/1/zip", destination, 100000)
+        self.assertEqual(destination.read_bytes(), archive)
+        self.assertEqual(len(signed_responses), 2)
+        self.assertEqual(origin_requests, 3)
+        self.assertEqual([item.args for item in sleep.call_args_list], [(1,), (2,)])
+
+    def test_download_rejects_malformed_zip_after_retries(self):
+        api = maintenance_module.Api(self.config["api_url"], self.token)
+        destination = self.root / "artifact.zip"
+        seen = []
+        target = "https://gitea.example.test/api/v1/repos/team/project/actions/artifacts/1/zip/raw?sig=test"
+
+        class Response(io.BytesIO):
+            headers = {"Content-Length": "9"}
+
+        class Opener:
+            def open(self, request, timeout):
+                seen.append(request)
+                if not isinstance(request, str):
+                    raise maintenance_module.urllib.error.HTTPError(request.full_url, 302, "Found", {"Location": target}, None)
+                return Response(b"not a zip")
+
+        with patch.object(maintenance_module.urllib.request, "build_opener", return_value=Opener()), \
+             patch.object(maintenance_module.time, "sleep") as sleep:
+            with self.assertRaisesRegex(RuntimeError, "incomplete after retries"):
+                api.download("repos/team/project/actions/artifacts/1/zip", destination, 9)
+        self.assertFalse(destination.exists())
+        self.assertEqual(sum(not isinstance(item, str) for item in seen), 3)
+        self.assertEqual([item.args for item in sleep.call_args_list], [(1,), (2,)])
+
+    def test_download_rejects_valid_zip_when_its_size_differs_from_metadata(self):
+        api = maintenance_module.Api(self.config["api_url"], self.token)
+        destination = self.root / "artifact.zip"
+        archive = zip_bytes()
+        seen = []
+        target = "https://gitea.example.test/api/v1/repos/team/project/actions/artifacts/1/zip/raw?sig=test"
+
+        class Opener:
+            def open(self, request, timeout):
+                seen.append(request)
+                if not isinstance(request, str):
+                    raise maintenance_module.urllib.error.HTTPError(request.full_url, 302, "Found", {"Location": target}, None)
+                return io.BytesIO(archive)
+
+        with patch.object(maintenance_module.urllib.request, "build_opener", return_value=Opener()), \
+             patch.object(maintenance_module.time, "sleep") as sleep:
+            with self.assertRaisesRegex(RuntimeError, "incomplete after retries"):
+                api.download("repos/team/project/actions/artifacts/1/zip", destination, len(archive) + 1)
+        self.assertFalse(destination.exists())
+        self.assertEqual(sum(not isinstance(item, str) for item in seen), 3)
+        self.assertEqual([item.args for item in sleep.call_args_list], [(1,), (2,)])
 
     def test_pending_chunk_cleanup_requires_old_terminal_master_run(self):
         instance = self.maintenance({"versions": [], "current": None, "attempts": [{"run_id": 4}]})

scripts/test_gitea_cache_snapshot.py

@@ -37,6 +37,16 @@ def object_path(character: str) -> str:
     return f"objects/{key[0]}/{key[1]}/{key}"
 
 
+def cache_object(entries, *, mode=0o100644) -> bytes:
+    output = io.BytesIO()
+    with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_STORED) as bundle:
+        for name, contents in entries:
+            info = zipfile.ZipInfo(name)
+            info.external_attr = mode << 16
+            bundle.writestr(info, contents)
+    return output.getvalue()
+
+
 class SnapshotMergeTest(unittest.TestCase):
     def setUp(self) -> None:
         self.temporary_directory = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
@@ -207,6 +217,42 @@ class SnapshotMergeTest(unittest.TestCase):
         with self.assertRaisesRegex(snapshot.SnapshotError, "conflicting content"):
             self.merge(inputs)
 
+    def test_deduplicates_cache_zip_member_order_preserving_payload_and_newest_touch(self) -> None:
+        path = object_path("a")
+        entries = [("lib.rlib", b"compiled"), ("lib.rmeta", b"metadata"), ("stderr", b"warning")]
+        first = cache_object(entries)
+        second = cache_object(list(reversed(entries)))
+        self.assertEqual(len(first), len(second))
+        self.assertNotEqual(hashlib.sha256(first).digest(), hashlib.sha256(second).digest())
+        inputs = [
+            self.archive("smoke", "smoke", 45, [(path, first, 100)]),
+            self.archive("lane-1", "lane-1", 45, [(path, second, 300)]),
+            self.archive("lane-2", "lane-2", 45, [(path, second, 200)]),
+        ]
+        result = self.merge(inputs)
+        output = self.root / "merged" / path
+        self.assertEqual((result.object_count, result.total_bytes), (1, len(first)))
+        self.assertEqual(output.read_bytes(), first)
+        self.assertEqual(output.stat().st_mtime_ns, 300)
+
+    def test_rejects_cache_zip_payload_or_permissions_conflicts(self) -> None:
+        path = object_path("a")
+        first = cache_object([("lib.rlib", b"one"), ("stderr", b"err")])
+        variants = {
+            "payload": cache_object([("stderr", b"err"), ("lib.rlib", b"two")]),
+            "permissions": cache_object([("stderr", b"err"), ("lib.rlib", b"one")], mode=0o100755),
+        }
+        for name, second in variants.items():
+            with self.subTest(name=name):
+                self.assertEqual(len(first), len(second))
+                inputs = [
+                    self.archive("first", "smoke", 45, [(path, first, 100)]),
+                    self.archive("second", "lane-1", 45, [(path, second, 200)]),
+                ]
+                with self.assertRaisesRegex(snapshot.SnapshotError, "conflicting content"):
+                    self.merge(inputs)
+                self.assertFalse((self.root / "merged").exists())
+
     def test_rejects_delta_that_conflicts_with_inherited_key(self) -> None:
         path = object_path("f")
         self.base_object(path, b"base", 1)