复用 master CI 产物自动更新编译缓存

仅由 master CI 导出新增缓存对象，宿主合并去重并组装限额快照，避免重复预热编译。
跟踪任务领取与最终上报，仅在安全空闲时切换镜像，移除管理员 API 依赖。
排除取消及不完整产物，定向清理过期镜像、归档和中断上传残块。
同步部署文档、共享记忆及缓存发布、合并和清理的定向测试。

scripts/ci-rust-cache.sh

@@ -1,5 +1,5 @@
 #!/usr/bin/env bash
-# 仅消费镜像内的可信快照；所有写入留在当前容器的可写层。
+# 消费镜像内可信快照；写入留在任务容器，master push 可在任务结束后导出。
 set -euo pipefail
 
 cache_root="${GENARRATIVE_CI_RUST_CACHE_ROOT:-/opt/genarrative-ci/rust-cache}"
@@ -22,7 +22,7 @@ configure_local_cache() {
 case "${1:-}" in
   prepare)
     : "${GITHUB_ENV:?GITHUB_ENV is required}"
-    printf 'RUSTC_WRAPPER=\nCARGO_BUILD_RUSTC_WRAPPER=\nGENARRATIVE_CI_RUST_CACHE_STATE=\n' >> "${GITHUB_ENV}"
+    printf 'RUSTC_WRAPPER=\nCARGO_BUILD_RUSTC_WRAPPER=\nGENARRATIVE_CI_RUST_CACHE_STATE=\nGENARRATIVE_CI_RUST_CACHE_EXPORT_READY=\n' >> "${GITHUB_ENV}"
     fallback() { printf '[rust-cache] mode=direct reason=%s\n' "$1"; exit 0; }
     [[ -x "${cache_binary}" && -d "${cache_root}/objects" && -f "${cache_root}/rustc.txt" && -f "${cache_root}/source-commit.txt" && -f "${cache_root}/workspace.txt" ]] \
       || fallback snapshot-unavailable
@@ -44,6 +44,10 @@ case "${1:-}" in
       fallback wrapper-probe-failed
     fi
     script_path="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd -P)/$(basename "${BASH_SOURCE[0]}")"
+    if [[ "${GITHUB_EVENT_NAME:-}" == push && "${GITHUB_REF:-}" == refs/heads/master ]]; then
+      # 只记 key/大小/时间，不读取对象正文；PR 没有扫描与打包开销。
+      python3 "$(dirname "${script_path}")/export-gitea-rust-cache.py" baseline "${state}/baseline.json"
+    fi
     # wrapper 路径也参与 Rust cache key；固定容器内路径，隔离由 job 容器保证。
     wrapper_path="${cache_root}/rustc-wrapper"
     printf '#!/usr/bin/env bash\nexec bash %q "$@"\n' "${script_path}" > "${wrapper_path}"
@@ -59,7 +63,13 @@ case "${1:-}" in
     if [[ -n "${state}" && -d "${state}" ]]; then
       configure_local_cache
       timeout --kill-after=2 5 "${cache_binary}" --show-stats || true
-      timeout --kill-after=2 5 "${cache_binary}" --stop-server >/dev/null 2>&1 || true
+      # 只有成功停服后才允许读取对象；失败不能把仍在写入的目录发布成完整快照。
+      if timeout --kill-after=2 15 "${cache_binary}" --stop-server >/dev/null 2>&1; then
+        if [[ "${GITHUB_EVENT_NAME:-}" == push && "${GITHUB_REF:-}" == refs/heads/master && ! -f "${state}/disabled" && -f "${state}/baseline.json" ]]; then
+          mv -- "${state}/baseline.json" "${cache_root}/export-baseline.json"
+          printf 'GENARRATIVE_CI_RUST_CACHE_EXPORT_READY=1\n' >> "${GITHUB_ENV}"
+        fi
+      fi
       rm -rf -- "${state}"
     else
       printf '[rust-cache] mode=direct\n'

scripts/ci-rust-cache.test.mjs

@@ -35,7 +35,8 @@ function fixture(t) {
     `#!/bin/bash
 set -eu
 case "$1" in
-  --stop-server|--show-stats) exit 0 ;;
+  --stop-server) exit "\${STOP_FAILURE:-0}" ;;
+  --show-stats) exit 0 ;;
 esac
 if [[ "$*" == *-vV ]]; then
   [[ "\${PROBE_FAILURE:-}" != 1 ]] || exit 1
@@ -215,3 +216,25 @@ linuxTest('lost local cache state still invokes the real compiler', (t) => {
   const f = fixture(t);
   assert.equal(f.run(['/bin/bash', '-c', 'exit 43']).status, 43);
 });
+
+linuxTest('only a cleanly stopped master cache can export objects', (t) => {
+  const master = {
+    GITHUB_EVENT_NAME: 'push',
+    GITHUB_REF: 'refs/heads/master',
+  };
+  for (const stopFailure of ['0', '1']) {
+    const f = fixture(t);
+    const prepared = f.run(['prepare'], master);
+    assert.equal(prepared.status, 0, prepared.stderr);
+    const result = f.run(['report'], {
+      ...f.preparedEnv(),
+      ...master,
+      STOP_FAILURE: stopFailure,
+    });
+    assert.equal(result.status, 0, result.stderr);
+    assert.equal(
+      f.preparedEnv().GENARRATIVE_CI_RUST_CACHE_EXPORT_READY,
+      stopFailure === '0' ? '1' : '',
+    );
+  }
+});

scripts/export-gitea-rust-cache.py

@@ -0,0 +1,274 @@
+#!/usr/bin/env python3
+"""只把 master push 的任务内 sccache 对象导出为 Gitea Actions 产物。"""
+
+import base64
+from datetime import datetime, timedelta, timezone
+import hashlib
+import io
+import json
+import os
+from pathlib import Path
+import re
+import stat
+import subprocess
+import sys
+import tarfile
+import tempfile
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+import zipfile
+
+
+MAX_OBJECT_BYTES = 4 * 1024**3
+CHUNK_BYTES = 8 * 1024**2
+JOBS = {
+    "ai-game-creator-shell-rust-lane-1",
+    "ai-game-creator-shell-rust-lane-2",
+    "ai-game-creator-shell-rust-smoke",
+    "ai-game-creator-shell-rust-crates",
+    "backend-tests",
+    "native-shell-tests",
+}
+
+
+def object_path(name):
+    parts = name.split("/")
+    return (
+        len(parts) == 3
+        and re.fullmatch(r"[a-f0-9]{64}", parts[2]) is not None
+        and parts[0] == parts[2][0]
+        and parts[1] == parts[2][1]
+    )
+
+
+class HashingReader:
+    def __init__(self, stream):
+        self.stream = stream
+        self.digest = hashlib.sha256()
+
+    def read(self, size=-1):
+        data = self.stream.read(size)
+        self.digest.update(data)
+        return data
+
+
+def scan_objects(root):
+    objects = root / "objects"
+    if objects.is_symlink() or not objects.is_dir():
+        raise ValueError("cache object directory is unavailable")
+    candidates = []
+    for directory, directories, files in os.walk(objects, followlinks=False):
+        directories[:] = [
+            name for name in directories if not (Path(directory) / name).is_symlink()
+        ]
+        for filename in files:
+            path = Path(directory) / filename
+            relative = path.relative_to(objects).as_posix()
+            info = path.lstat()
+            if object_path(relative) and stat.S_ISREG(info.st_mode):
+                candidates.append((path, relative, info))
+    return candidates
+
+
+def save_baseline(root, destination):
+    # sccache 命中会更新 mtime，不能把时间变化当成内容变化，否则又变成全量上传。
+    baseline = {
+        relative: {"size": info.st_size, "mtime_ns": info.st_mtime_ns}
+        for _, relative, info in scan_objects(root)
+    }
+    destination.write_text(json.dumps(baseline), encoding="utf-8")
+
+
+def pack_snapshot(root, destination, metadata, baseline, limit=MAX_OBJECT_BYTES):
+    """仅归档新 key；命中的已有对象只传递新近使用时间，不传输对象内容。"""
+    candidates = scan_objects(root)
+    candidates.sort(key=lambda item: (-item[2].st_mtime_ns, item[1]))
+    entries = []
+    touched = []
+    total = 0
+    with zipfile.ZipFile(destination, "w", compression=zipfile.ZIP_STORED) as bundle, \
+            bundle.open("snapshot.tar", "w", force_zip64=True) as tar_stream, \
+            tarfile.open(fileobj=tar_stream, mode="w|", format=tarfile.USTAR_FORMAT) as archive:
+        for path, relative, info in candidates:
+            previous = baseline.get(relative)
+            if previous is not None and previous["size"] == info.st_size:
+                if previous["mtime_ns"] != info.st_mtime_ns:
+                    touched.append({"path": "objects/" + relative, "mtime_ns": info.st_mtime_ns})
+                continue
+            if info.st_size <= 0 or total + info.st_size > limit:
+                continue
+            # report 已成功停止 daemon；打开后再核实对象，避免导出变化中的文件。
+            descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
+            with os.fdopen(descriptor, "rb") as stream:
+                actual = os.fstat(stream.fileno())
+                if not stat.S_ISREG(actual.st_mode) or (
+                    actual.st_size, actual.st_mtime_ns, actual.st_ino
+                ) != (info.st_size, info.st_mtime_ns, info.st_ino):
+                    raise ValueError("cache object changed during export")
+                member = tarfile.TarInfo("objects/" + relative)
+                member.size = info.st_size
+                member.mode = 0o644
+                member.mtime = int(info.st_mtime)
+                reader = HashingReader(stream)
+                archive.addfile(member, reader)
+                entries.append({
+                    "path": member.name,
+                    "size": info.st_size,
+                    "sha256": reader.digest.hexdigest(),
+                    "mtime_ns": info.st_mtime_ns,
+                })
+                total += info.st_size
+        manifest = dict(metadata, schema=1, mode="delta", objects=entries, touched=touched)
+        data = json.dumps(manifest, ensure_ascii=False, sort_keys=True).encode("utf-8")
+        member = tarfile.TarInfo("manifest.json")
+        member.size = len(data)
+        member.mode = 0o644
+        archive.addfile(member, io.BytesIO(data))
+    return len(entries), total
+
+
+class NoRedirect(urllib.request.HTTPRedirectHandler):
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        # runtime token 只允许发往明确配置的 Gitea 地址。
+        return None
+
+
+def request(method, url, token, data=None, headers=None, attempts=3):
+    opener = urllib.request.build_opener(NoRedirect())
+    for attempt in range(attempts):
+        req = urllib.request.Request(url, data=data, method=method, headers={
+            **({"Authorization": "Bearer " + token} if token else {}),
+            "Content-Type": "application/json",
+            **(headers or {}),
+        })
+        try:
+            with opener.open(req, timeout=120) as response:
+                body = response.read(1024 * 1024)
+                return json.loads(body) if body else {}
+        except urllib.error.HTTPError as error:
+            if error.code not in (408, 429, 500, 502, 503, 504) or attempt == attempts - 1:
+                raise RuntimeError(f"artifact {method} failed: HTTP {error.code}") from None
+        except (urllib.error.URLError, TimeoutError, ConnectionError):
+            if attempt == attempts - 1:
+                raise RuntimeError(f"artifact {method} connection failed") from None
+        time.sleep(2 ** attempt)
+
+
+def artifact_base_url(env):
+    # .runner.address 可指向仅支持 RPC 的领取网关，不能用 GITHUB_SERVER_URL。
+    repository = env["GITHUB_REPOSITORY"]
+    clone_url = env["GENARRATIVE_GITEA_REPOSITORY_URL"]
+    suffix = "/" + repository + ".git"
+    parsed = urllib.parse.urlsplit(clone_url)
+    if (
+        parsed.scheme not in ("http", "https")
+        or not parsed.netloc
+        or parsed.username or parsed.password or parsed.query or parsed.fragment
+        or not parsed.path.endswith(suffix)
+    ):
+        raise ValueError("a credential-free HTTP Gitea repository URL is required")
+    return urllib.parse.urlunsplit((
+        parsed.scheme, parsed.netloc, parsed.path[:-len(suffix)], "", ""
+    ))
+
+
+def upload_snapshot(path, name, run_id, base_url, token):
+    """Gitea 1.26.4 原生 v4：只有完成全部块并校验 SHA256 后才发布。"""
+    producer_match = re.fullmatch(r"rust-cache-v1-([a-z0-9-]+)-attempt-(0|[1-9][0-9]*)", name)
+    if not producer_match or producer_match[1] not in JOBS:
+        raise ValueError("unexpected cache artifact name")
+    producer = f"{producer_match[1]}:{producer_match[2]}"
+    endpoint = f"{base_url}/twirp/github.actions.results.api.v1.ArtifactService"
+    # Gitea 从任务凭据确定 job，只要求请求中的 run ID 与任务所属 run 一致。
+    identity = {"workflowRunBackendId": str(run_id), "name": name}
+    created = request("POST", endpoint + "/CreateArtifact", token, json.dumps({
+        **identity, "version": 4,
+        # Gitea 将剩余小时向下取整成天；多给一天，服务端才实际保留至少七天。
+        "expiresAt": (datetime.now(timezone.utc) + timedelta(days=8)).isoformat(),
+    }).encode())
+    received = urllib.parse.urlsplit(created.get("signedUploadUrl", ""))
+    expected_path = urllib.parse.urlsplit(endpoint).path + "/UploadArtifact"
+    if not created.get("ok") or received.path != expected_path or not received.query:
+        raise ValueError("unexpected artifact upload path")
+    # 服务可能返回外部 AppURL；保留签名，但只连接配置的内部 Gitea 地址。
+    upload_url = endpoint + "/UploadArtifact?" + received.query
+    size = path.stat().st_size
+    digest = hashlib.sha256()
+    blocks = []
+    with path.open("rb") as source:
+        while chunk := source.read(CHUNK_BYTES):
+            # 双层编码中的内层保留所属任务；宿主只回收带此前缀的过期未完成块。
+            block = base64.b64encode(
+                f"genarrative-rust-cache-v1:{producer}:{len(blocks):08d}".encode()
+            ).decode()
+            blocks.append(block)
+            digest.update(chunk)
+            request("PUT", upload_url + "&" + urllib.parse.urlencode({
+                "comp": "block", "blockid": block,
+            }), None, chunk, {
+                "Content-Type": "application/octet-stream",
+            })
+    blocklist = "<BlockList>" + "".join(f"<Latest>{block}</Latest>" for block in blocks) + "</BlockList>"
+    request("PUT", upload_url + "&comp=blocklist", None, blocklist.encode(), {
+        "Content-Type": "application/xml",
+    })
+    # Finalize 会消耗服务端块列表；响应不确定时不盲目重试、也不宣告成功。
+    finalized = request("POST", endpoint + "/FinalizeArtifact", token, json.dumps({
+        **identity, "size": str(size), "hash": "sha256:" + digest.hexdigest(),
+    }).encode(), attempts=1)
+    if not finalized.get("ok"):
+        raise ValueError("artifact was not finalized")
+
+
+def export(env):
+    # 双重限定：手工调用、PR 和其它分支不会扫描、打包或上传对象。
+    if env.get("GITHUB_EVENT_NAME") != "push" or env.get("GITHUB_REF") != "refs/heads/master":
+        return
+    if env.get("GENARRATIVE_CI_RUST_CACHE_EXPORT_READY") != "1":
+        print("[rust-cache] export skipped: cache daemon did not stop cleanly")
+        return
+    job = env.get("GITHUB_JOB")
+    if job not in JOBS:
+        raise ValueError("unexpected Rust cache job")
+    sha = env["GITHUB_SHA"]
+    if not re.fullmatch(r"[a-f0-9]{40}", sha):
+        raise ValueError("a complete source SHA is required")
+    run_id, attempt = int(env["GITHUB_RUN_ID"]), int(env["GITHUB_RUN_ATTEMPT"])
+    if run_id <= 0 or attempt < 0:
+        raise ValueError("invalid run identity")
+    token = env.get("ACTIONS_RUNTIME_TOKEN") or env["GENARRATIVE_GITEA_TOKEN"]
+    base_url = artifact_base_url(env)
+    root = Path(env.get("GENARRATIVE_CI_RUST_CACHE_ROOT", "/opt/genarrative-ci/rust-cache"))
+    baseline_path = root / "export-baseline.json"
+    baseline = json.loads(baseline_path.read_text(encoding="utf-8"))
+    rustc = subprocess.check_output(["rustc", "-vV"], text=True)
+    workspace = str(Path.cwd().resolve())
+    if rustc != (root / "rustc.txt").read_text() or workspace != (root / "workspace.txt").read_text().strip():
+        raise ValueError("cache provenance changed after prepare")
+    metadata = {
+        "repository": env["GITHUB_REPOSITORY"],
+        "run_id": run_id, "run_attempt": attempt, "job": job, "source_sha": sha,
+        "rustc": rustc, "workspace": workspace,
+        "inherited_source_sha": (root / "source-commit.txt").read_text().strip(),
+        "sccache_version": subprocess.check_output([str(root / "sccache"), "--version"], text=True).strip(),
+    }
+    if (root / "base-image.txt").is_file():
+        metadata["base_image"] = (root / "base-image.txt").read_text().strip()
+    name = f"rust-cache-v1-{job}-attempt-{attempt}"
+    try:
+        with tempfile.TemporaryDirectory(prefix="ci-rust-cache-export-") as directory:
+            path = Path(directory) / "snapshot.zip"
+            count, size = pack_snapshot(root, path, metadata, baseline)
+            upload_snapshot(path, name, run_id, base_url, token)
+    finally:
+        baseline_path.unlink(missing_ok=True)
+    print(f"[rust-cache] artifact={name} objects={count} bytes={size} complete=true")
+
+
+if __name__ == "__main__":
+    if len(sys.argv) == 3 and sys.argv[1] == "baseline":
+        save_baseline(Path(os.environ.get("GENARRATIVE_CI_RUST_CACHE_ROOT", "/opt/genarrative-ci/rust-cache")), Path(sys.argv[2]))
+    else:
+        export(os.environ)

scripts/gitea-runner-fetch-gate.py

@@ -3,6 +3,8 @@
 
 import http.client
 import http.server
+import gzip
+import io
 import json
 import os
 from pathlib import Path
@@ -10,12 +12,90 @@ import socketserver
 import threading
 import time
 import urllib.parse
+import zlib
 
 MAX_BODY = 32 * 1024 * 1024
 HOP_HEADERS = {
     "connection", "keep-alive", "proxy-authenticate", "proxy-authorization",
     "te", "trailer", "transfer-encoding", "upgrade", "host", "content-length",
 }
+RPC_PREFIX = "/api/actions/runner.v1.RunnerService/"
+
+
+def protobuf_fields(data):
+    """只解码已核对的 actions-proto-go v0.4.1 字段，不引入 protobuf 运行时。"""
+    fields = {}
+    offset = 0
+
+    def varint():
+        nonlocal offset
+        value = 0
+        for shift in range(0, 70, 7):
+            if offset >= len(data):
+                raise ValueError("truncated protobuf")
+            byte = data[offset]
+            offset += 1
+            value |= (byte & 127) << shift
+            if byte < 128:
+                return value
+        raise ValueError("invalid protobuf varint")
+
+    while offset < len(data):
+        tag = varint()
+        number, wire = tag >> 3, tag & 7
+        if not number:
+            raise ValueError("invalid protobuf tag")
+        if wire == 0:
+            value = varint()
+        elif wire in (1, 2, 5):
+            length = varint() if wire == 2 else (8 if wire == 1 else 4)
+            if offset + length > len(data):
+                raise ValueError("truncated protobuf field")
+            value = data[offset:offset + length]
+            offset += length
+        else:
+            raise ValueError("unsupported protobuf wire type")
+        fields.setdefault(number, []).append(value)
+    return fields
+
+
+def single(fields, number, default=None):
+    values = fields.get(number, [])
+    if len(values) > 1:
+        raise ValueError("ambiguous tracking field")
+    return values[0] if values else default
+
+
+def rpc_fields(body, headers):
+    encoding = headers.get("Content-Encoding", "identity").lower()
+    if encoding == "gzip":
+        with gzip.GzipFile(fileobj=io.BytesIO(body)) as stream:
+            body = stream.read(MAX_BODY + 1)
+    elif encoding != "identity":
+        raise ValueError("unsupported RPC encoding")
+    if len(body) > MAX_BODY:
+        raise ValueError("decoded RPC too large")
+    if headers.get("Content-Type", "").split(";", 1)[0] != "application/proto":
+        raise ValueError("task tracking requires Connect protobuf")
+    return protobuf_fields(body)
+
+
+def task_state(fields):
+    nested = single(fields, 1)
+    if not isinstance(nested, bytes):
+        raise ValueError("missing task state")
+    state = protobuf_fields(nested)
+    task_id = positive_id(single(state, 1))
+    result = single(state, 2, 0)
+    if type(result) is not int or result not in range(5):
+        raise ValueError("unknown task result")
+    return task_id, result
+
+
+def positive_id(value):
+    if type(value) is not int or not 0 < value < 2 ** 63:
+        raise ValueError("invalid task ID")
+    return str(value)
 
 
 class Gate:
@@ -26,7 +106,19 @@ class Gate:
         self.inflight = 0
         self.last_fetch_peer = None
         self.last_fetch_at = None
+        self.tasks = {}
         self.uncertain = (self.directory / "uncertain").exists()
+        try:
+            if (self.directory / "tasks.json").exists():
+                tasks = json.loads((self.directory / "tasks.json").read_text())
+                if (not isinstance(tasks, dict) or any(
+                        positive_id(int(key)) != key or type(value) is not bool
+                        for key, value in tasks.items())):
+                    raise ValueError("invalid task ledger")
+                self.tasks = tasks
+        except (ValueError, TypeError, OSError):
+            self.uncertain = True
+            self.mark("uncertain")
         if (self.directory / "inflight").exists():
             self.uncertain = True
             self.mark("uncertain")
@@ -48,9 +140,71 @@ class Gate:
 
     def snapshot(self):
         return {"paused": self.paused, "inflight": self.inflight,
+                "active_tasks": len(self.tasks), "task_ids": sorted(self.tasks),
                 "uncertain": self.uncertain, "last_fetch_peer": self.last_fetch_peer,
                 "last_fetch_at": self.last_fetch_at}
 
+    def save_tasks(self):
+        temporary = self.directory / "tasks.json.tmp"
+        with temporary.open("w", encoding="ascii") as stream:
+            json.dump(self.tasks, stream, sort_keys=True)
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temporary, self.directory / "tasks.json")
+        self.sync_directory()
+
+    def fail_closed(self):
+        with self.lock:
+            self.uncertain = self.paused = True
+            self.mark("uncertain")
+
+    def assigned(self, fields):
+        task = single(fields, 1)
+        if task is None:
+            return
+        if not isinstance(task, bytes):
+            raise ValueError("invalid fetched task")
+        task_id = positive_id(single(protobuf_fields(task), 1))
+        with self.lock:
+            self.tasks[task_id] = False
+            # 必须先持久化，再把领取结果交给 runner；仅保存 ID，不保存 secrets。
+            self.save_tasks()
+
+    def reported(self, method, request, response):
+        if method == "UpdateLog":
+            task_id = positive_id(single(request, 1))
+            index = single(request, 2, 0)
+            ack = single(response, 1, 0)
+            no_more = single(request, 4, 0)
+            if (type(index) is not int or type(ack) is not int
+                    or index < 0 or ack < 0 or no_more not in (0, 1)):
+                raise ValueError("invalid log acknowledgement")
+            finalized = no_more == 1 and ack == index + len(request.get(3, []))
+        else:
+            task_id, result = task_state(request)
+            response_id, response_result = task_state(response)
+            if response_id != task_id:
+                raise ValueError("mismatched task acknowledgement")
+            # 取消响应可能出现在任务执行途中，必须等 runner 自己报告终态。
+            finalized = result != 0 and response_result != 0
+            output_keys = {single(protobuf_fields(entry), 1, b"")
+                           for entry in request.get(2, [])}
+            finalized = finalized and output_keys.issubset(set(response.get(2, [])))
+        with self.lock:
+            if task_id not in self.tasks:
+                # 客户端可能未读到已成功发出的终态响应而重试；v2.0.0 仅在
+                # executor 清理后的 Close 中发送终态，不为幂等重报增加墓碑账本。
+                if method == "UpdateTask" and finalized:
+                    return
+                raise ValueError("report for untracked task; idle bootstrap required")
+            if method == "UpdateLog" and finalized:
+                self.tasks[task_id] = True
+                self.save_tasks()
+            elif method == "UpdateTask" and finalized and self.tasks[task_id]:
+                # act_runner Reporter.Close 在 executor 清理之后先封存日志再报终态。
+                del self.tasks[task_id]
+                self.save_tasks()
+
     def record_fetch(self, peer):
         with self.lock:
             self.last_fetch_peer = peer
@@ -142,6 +296,7 @@ class Proxy(http.server.BaseHTTPRequestHandler):
         pass  # 不输出 RPC 认证头、请求内容或带认证信息的 URL。
 
     def reply(self, status, body, headers=()):
+        delivered = False
         try:
             self.send_response(status)
             excluded = HOP_HEADERS | {
@@ -155,9 +310,12 @@ class Proxy(http.server.BaseHTTPRequestHandler):
             self.send_header("Connection", "close")
             self.end_headers()
             self.wfile.write(body)
+            self.wfile.flush()
+            delivered = True
         except (OSError, ValueError):
             pass
         self.close_connection = True
+        return delivered
 
     def do_POST(self):
         path = urllib.parse.urlsplit(self.path)
@@ -170,7 +328,8 @@ class Proxy(http.server.BaseHTTPRequestHandler):
         except (ValueError, OSError):
             self.reply(400, b"invalid request body\n")
             return
-        is_fetch = path.path.endswith("/FetchTask")
+        method = path.path.removeprefix(RPC_PREFIX) if path.path.startswith(RPC_PREFIX) else ""
+        is_fetch = method == "FetchTask"
         if is_fetch:
             # 包括暂停时被拒绝的请求；只记录网络来源与时间以证明 daemon 路由。
             self.server.gate.record_fetch(self.client_address[0])
@@ -209,12 +368,23 @@ class Proxy(http.server.BaseHTTPRequestHandler):
                 raise http.client.IncompleteRead(bytes(result), response.length)
             completed = True
             if is_fetch:
+                try:
+                    if oversized or response.status != 200:
+                        raise ValueError("unknown FetchTask result")
+                    self.server.gate.assigned(rpc_fields(bytes(result), response.headers))
+                except (ValueError, TypeError, OSError, EOFError, zlib.error):
+                    self.server.gate.fail_closed()
                 self.server.gate.leave(True)
                 is_fetch = False
             if oversized:
                 self.reply(502, b"upstream response too large\n")
             else:
-                self.reply(response.status, bytes(result), response.getheaders())
+                # 日志封存先记账再返回，避免 runner 立即发终态时抢先读到旧账本。
+                if method == "UpdateLog" and response.status == 200:
+                    self.observe_report(method, body, bytes(result), response.headers)
+                delivered = self.reply(response.status, bytes(result), response.getheaders())
+                if method == "UpdateTask" and response.status == 200 and delivered:
+                    self.observe_report(method, body, bytes(result), response.headers)
         except (OSError, http.client.HTTPException, ValueError):
             self.reply(502, b"runner upstream unavailable\n")
         finally:
@@ -223,6 +393,13 @@ class Proxy(http.server.BaseHTTPRequestHandler):
             if is_fetch:
                 self.server.gate.leave(completed)
 
+    def observe_report(self, method, request, response, headers):
+        try:
+            self.server.gate.reported(method, rpc_fields(request, self.headers),
+                                      rpc_fields(response, headers))
+        except (ValueError, TypeError, OSError, EOFError, zlib.error):
+            self.server.gate.fail_closed()
+
 
 class Control(socketserver.StreamRequestHandler):
     def handle(self):

scripts/gitea_cache_upload_cleanup.py

@@ -0,0 +1,115 @@
+"""定向回收 Gitea 1.26.4 不会自动过期的本仓库缓存上传块。"""
+
+import base64
+import binascii
+import math
+from pathlib import Path
+import re
+import stat
+
+
+JOBS = {
+    "ai-game-creator-shell-rust-lane-1",
+    "ai-game-creator-shell-rust-lane-2",
+    "ai-game-creator-shell-rust-smoke",
+    "ai-game-creator-shell-rust-crates",
+    "backend-tests",
+    "native-shell-tests",
+}
+CHUNK = re.compile(r"block-([1-9][0-9]*)-([1-9][0-9]*)-(0|[1-9][0-9]*)-([A-Za-z0-9_-]+={0,2})")
+BLOCKLIST = re.compile(r"([1-9][0-9]*)-([1-9][0-9]*)-blocklist")
+MARKER = re.compile(r"genarrative-rust-cache-v1:([a-z0-9-]+):(0|[1-9][0-9]*):([0-9]{8})")
+
+
+def decode_owner(encoded):
+    """Gitea 文件名编码为 URL-base64(生产者上传的标准 base64 blockid)。"""
+    try:
+        block_id = base64.b64decode(encoded, altchars=b"-_", validate=True)
+        if base64.urlsafe_b64encode(block_id).decode() != encoded:
+            return None
+        marker = base64.b64decode(block_id, validate=True)
+        if base64.b64encode(marker) != block_id:
+            return None
+        match = MARKER.fullmatch(marker.decode("ascii"))
+        if match and match[1] in JOBS:
+            return match[1], int(match[2])
+    except (binascii.Error, UnicodeError, ValueError):
+        pass
+    return None
+
+
+def _unchanged_old_regular(path, previous, cutoff):
+    try:
+        current = path.lstat()
+    except FileNotFoundError:
+        return False
+    return (
+        stat.S_ISREG(current.st_mode)
+        and current.st_mtime < cutoff
+        and (current.st_ino, current.st_size, current.st_mtime_ns)
+        == (previous.st_ino, previous.st_size, previous.st_mtime_ns)
+    )
+
+
+def cleanup_upload_chunks(storage_root: Path, eligible_runs: set[int], cutoff_timestamp: float) -> int:
+    """调用方先用仓库 API 确认这些 master run 已结束超过保留期限。"""
+    storage_root = Path(storage_root)
+    if not storage_root.is_absolute() or storage_root.resolve(strict=True) != storage_root:
+        raise ValueError("artifact storage root must be an existing absolute real directory")
+    if not stat.S_ISDIR(storage_root.lstat().st_mode):
+        raise ValueError("artifact storage root must be a directory")
+    if not math.isfinite(cutoff_timestamp) or cutoff_timestamp <= 0:
+        raise ValueError("invalid artifact cleanup cutoff")
+    if any(type(run_id) is not int or run_id <= 0 for run_id in eligible_runs):
+        raise ValueError("invalid eligible run ID")
+    temporary = storage_root / "tmp-upload"
+    if not temporary.exists() and not temporary.is_symlink():
+        return 0
+    if temporary.is_symlink() or not temporary.is_dir():
+        raise ValueError("artifact temporary path must be a real directory")
+
+    deleted = 0
+    for run_id in sorted(eligible_runs):
+        directory = temporary / f"run-{run_id}-v4"
+        if directory.is_symlink() or not directory.is_dir():
+            continue
+        groups = {}
+        unknown = False
+        for path in directory.iterdir():
+            chunk = CHUNK.fullmatch(path.name)
+            blocklist = BLOCKLIST.fullmatch(path.name)
+            match = chunk or blocklist
+            if match and int(match[1]) == run_id:
+                artifact_id = int(match[2])
+                owner = decode_owner(chunk[4]) if chunk else None
+                owned = owner is not None if chunk else True
+            else:
+                # 若仍能识别 artifact ID，仅保护该组；无法识别时保护整个 run。
+                prefix = re.match(rf"(?:block-)?{run_id}-([1-9][0-9]*)-", path.name)
+                if not prefix:
+                    unknown = True
+                    break
+                artifact_id, owner, owned = int(prefix[1]), None, False
+            group = groups.setdefault(artifact_id, {"files": [], "owners": set(), "safe": True})
+            try:
+                info = path.lstat()
+            except FileNotFoundError:
+                group["safe"] = False
+                continue
+            group["safe"] &= owned and stat.S_ISREG(info.st_mode) and info.st_mtime < cutoff_timestamp
+            if owner:
+                group["owners"].add(owner)
+            group["files"].append((path, info))
+        if unknown:
+            continue
+        for group in groups.values():
+            if not group["safe"] or len(group["owners"]) != 1:
+                continue
+            files = group["files"]
+            if not all(_unchanged_old_regular(path, info, cutoff_timestamp) for path, info in files):
+                continue
+            # 保留目录，不使用递归删除；没有所属块证明的孤立 blocklist 也不会删除。
+            for path, _ in files:
+                path.unlink()
+                deleted += 1
+    return deleted

scripts/maintain-gitea-rust-cache.py

@@ -9,13 +9,18 @@ import os
 from pathlib import Path
 import re
 import socket
+import shutil
 import subprocess
 import sys
+import tempfile
 import time
 import urllib.error
 import urllib.parse
 import urllib.request
 
+from gitea_cache_snapshot import ArtifactIdentity, ArtifactInput, merge_snapshots
+from gitea_cache_upload_cleanup import cleanup_upload_chunks
+
 
 IMAGE = re.compile(r"sha256:[0-9a-f]{64}\Z")
 SHA = re.compile(r"[0-9a-f]{40}\Z")
@@ -29,6 +34,17 @@ JOBS = {
     "Repository checks", "AI game creator shell web tests",
 }
 RUST_JOBS = {name for name in JOBS if "Rust" in name or name in {"Backend tests", "Native shell tests"}}
+RUST_JOB_IDS = {
+    "AI game creator shell Rust lane 1/2": "ai-game-creator-shell-rust-lane-1",
+    "AI game creator shell Rust lane 2/2": "ai-game-creator-shell-rust-lane-2",
+    "AI game creator shell Rust smoke": "ai-game-creator-shell-rust-smoke",
+    "AI game creator shell Rust crates": "ai-game-creator-shell-rust-crates",
+    "Backend tests": "backend-tests",
+    "Native shell tests": "native-shell-tests",
+}
+ARTIFACT_PREFIX = "rust-cache-v1-"
+MAX_DOWNLOAD = 4 * 1024 ** 3 + 129 * 1024 ** 2
+EXPORT_STEP = "Publish master Rust cache artifact"
 
 
 def now():
@@ -127,14 +143,14 @@ class Api:
             raise ValueError("api_url must use HTTPS")
         self.token_file = Path(token_file)
 
-    def request(self, path, body=None, raw=False):
+    def request(self, path, body=None, raw=False, method=None):
         token = self.token_file.read_text().strip()
         if not token or "\n" in token:
             raise ValueError("invalid token file")
         request = urllib.request.Request(
             self.url + "/" + path.lstrip("/"),
             data=None if body is None else json.dumps(body).encode(),
-            method="GET" if body is None else "PATCH",
+            method=method or ("GET" if body is None else "PATCH"),
             headers={"Authorization": "token " + token, "Content-Type": "application/json"},
         )
         try:
@@ -143,8 +159,41 @@ class Api:
             with opener.open(request, timeout=30) as response:
                 content = response.read().decode()
         except urllib.error.HTTPError as error:
+            error.close()
+            if method == "DELETE" and error.code == 404:
+                return None
             raise RuntimeError(f"Gitea API HTTP {error.code}") from None
-        return content if raw else json.loads(content)
+        return content if raw else (json.loads(content) if content else None)
+
+    def download(self, path, destination):
+        """REST V4 archive redirects to a signed URL; never forward the API token."""
+        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})
+        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):
+                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)
+        try:
+            with response, 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")
+                    out.write(chunk)
+        except Exception:
+            destination.unlink(missing_ok=True)
+            raise
 
     def pages(self, path, key):
         separator = "&" if "?" in path else "?"
@@ -175,9 +224,6 @@ class Maintenance:
         self.runner = config.get("runner_container", "gitea-runner")
         self.api = Api(config["api_url"], config["token_file"])
         self.repo_api = "repos/" + config["repository"]
-        self.runner_api = config["runner_api_path"].strip("/")
-        if not re.fullmatch(r"admin/actions/runners/\d+", self.runner_api):
-            raise ValueError("this service requires the global runner admin endpoint")
         self.state_path = self.root / "state.json"
         self.state = json.loads(self.state_path.read_text()) if self.state_path.exists() else {
             "versions": [], "current": None, "rollback": None, "candidate": None,
@@ -250,40 +296,149 @@ class Maintenance:
             command("bash", str(self.repo / "scripts" / script), *args, cwd=self.repo,
                     env=env, output=out, timeout=7200)
 
-    def build(self, sha, inputs):
+    def master_run(self, run):
+        return (run.get("path") == "project-ci.yml@refs/heads/master"
+                and run.get("event") == "push" and run.get("head_branch") == "master"
+                and SHA.fullmatch(run.get("head_sha", "")) is not None)
+
+    def source_run(self):
+        """Only complete exports from the latest eligible master run; never mix runs."""
+        current = self.version(self.state["current"])
+        current_inputs = (cache_inputs(command("git", "ls-tree", "-rz", current["source"], cwd=self.repo))
+                          if current.get("source") else None)
+        for run in self.api.pages(self.repo_api + "/actions/runs?branch=master&event=push", "workflow_runs"):
+            if (not self.master_run(run) or run.get("status") != "completed"
+                    or run.get("conclusion") not in {"success", "failure"}
+                    or run["id"] <= current.get("run_id", 0)):
+                continue
+            sha = run["head_sha"]
+            if sha == current.get("source"):
+                continue
+            ancestry = subprocess.run(["git", "merge-base", "--is-ancestor", sha, "FETCH_HEAD"],
+                                      cwd=self.repo, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            if ancestry.returncode:
+                continue
+            if current.get("source"):
+                ancestry = subprocess.run(["git", "merge-base", "--is-ancestor", current["source"], sha],
+                                          cwd=self.repo, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+                if ancestry.returncode:
+                    continue
+            inputs = cache_inputs(command("git", "ls-tree", "-rz", sha, cwd=self.repo))
+            if inputs == current_inputs:
+                continue
+            jobs = list(self.api.pages(self.repo_api + f'/actions/runs/{run["id"]}/jobs', "jobs"))
+            rust_jobs = [job for job in jobs if job["name"] in RUST_JOB_IDS]
+            if len(rust_jobs) != 6 or {job["name"] for job in rust_jobs} != RUST_JOBS:
+                continue
+            if any(job.get("status") != "completed" or job.get("conclusion") not in {"success", "failure"}
+                   or job.get("head_sha") != sha
+                   or not any(step.get("name") == EXPORT_STEP and step.get("conclusion") == "success"
+                              for step in job.get("steps", [])) for job in rust_jobs):
+                continue
+            artifacts = list(self.api.pages(self.repo_api + f'/actions/runs/{run["id"]}/artifacts', "artifacts"))
+            selected = []
+            images = set()
+            for job in rust_jobs:
+                name = f'{ARTIFACT_PREFIX}{RUST_JOB_IDS[job["name"]]}-attempt-{job["run_attempt"]}'
+                matches = [item for item in artifacts if item["name"] == name and not item.get("expired")
+                           and item.get("workflow_run", {}).get("id") == run["id"]]
+                if len(matches) != 1:
+                    break
+                content = self.api.request(self.repo_api + f'/actions/jobs/{job["id"]}/logs', raw=True)
+                if not re.search(re.escape(f"[rust-cache] artifact={name}")
+                                 + r" objects=\d+ bytes=\d+ complete=true", content):
+                    break
+                used = set(re.findall(r"(?m)^\S+ image: (sha256:[0-9a-f]{64})\s*$", content))
+                if len(used) != 1:
+                    break
+                images.update(used)
+                selected.append({"id": matches[0]["id"], "name": name,
+                                 "job": RUST_JOB_IDS[job["name"]], "attempt": job["run_attempt"]})
+            if len(selected) != 6 or len(images) != 1:
+                continue
+            source_image = images.pop()
+            if source_image not in {row["image"] for row in self.state["versions"]}:
+                continue
+            return {"run_id": run["id"], "source": sha, "inputs": inputs, "source_image": source_image,
+                    "exports": selected}
+        return None
+
+    def build(self, source):
+        """Assemble existing CI objects; no cargo warm-up or test execution."""
+        sha = source["source"]
+        command("git", "checkout", "--detach", "--force", sha, cwd=self.repo)
+        command("git", "clean", "-ffdx", cwd=self.repo)
         artifact = self.root / "artifacts" / sha
         artifact.mkdir(parents=True, exist_ok=True)
         build_log = artifact / "build.log"
         tag = "genarrative/gitea-project-ci:rust-cache-auto-" + sha
         base_tag = "genarrative/gitea-project-ci:base-auto-" + sha
-        self.state.setdefault("attempts", []).append({
-            "source": sha, "inputs": inputs, "tag": tag,
-            "base_tag": base_tag, "artifact": str(artifact),
-        })
+        attempt = {**source, "tag": tag, "base_tag": base_tag, "artifact": str(artifact)}
+        self.state.setdefault("attempts", []).append(attempt)
         self.save()
         env = {**os.environ, "GENARRATIVE_GITEA_RUNNER_CONTAINER": self.runner}
         env.pop("CI", None)
-        current = self.version(self.state["current"])
-        base = current["base"]
+        labels = self.image_info(source["source_image"])["Config"].get("Labels") or {}
+        inherited_source = labels.get("world.genarrative.ci.rust-cache-source")
+        base = labels.get("world.genarrative.ci.rust-cache-base")
+        if not IMAGE.fullmatch(base or "") or not SHA.fullmatch(inherited_source or ""):
+            raise RuntimeError("source image must contain a trusted cache snapshot")
         revision = command("bash", "scripts/gitea-ci-job-image.sh", "revision", cwd=self.repo).strip()
-        labels = self.image_info(base)["Config"].get("Labels") or {}
-        if labels.get("com.genarrative.ci.definition-sha256") != revision:
+        base_labels = self.image_info(base)["Config"].get("Labels") or {}
+        if base_labels.get("com.genarrative.ci.definition-sha256") != revision:
             env["GENARRATIVE_GITEA_CI_IMAGE_TAG"] = base_tag
             self.build_command(build_log, "gitea-ci-job-image.sh", "build", env=env)
             base = self.image_info(base_tag)["Id"]
             self.state.setdefault("bases", {})[base] = base_tag
             self.save()
-        self.build_command(build_log, "build-gitea-rust-cache.sh", base, tag, sha, env=env)
+        # 与旧缓存镜像分离；绝不把 Docker 可写层、源码或 target commit 成镜像。
+        self.docker("run", "--rm", "--network", "none", "--read-only", "--cap-drop=ALL",
+                    "--entrypoint", "bash", base, "-c", "test ! -e /opt/genarrative-ci/rust-cache")
+        with tempfile.TemporaryDirectory(prefix="assemble-", dir=artifact) as temporary:
+            work = Path(temporary)
+            inherited = work / "inherited"
+            inherited.mkdir()
+            container = self.docker("create", source["source_image"]).strip()
+            try:
+                self.docker("cp", container + ":/opt/genarrative-ci/rust-cache/.", str(inherited), timeout=600)
+            finally:
+                self.docker("rm", "--volumes", container)
+            inputs = []
+            for export in source["exports"]:
+                archive = work / (str(export["id"]) + ".zip")
+                self.api.download(self.repo_api + f'/actions/artifacts/{export["id"]}/zip', archive)
+                inputs.append(ArtifactInput(archive, ArtifactIdentity(
+                    self.config["repository"], source["run_id"], export["attempt"], export["job"], sha)))
+            snapshot = work / "snapshot"
+            merged = merge_snapshots(inputs, snapshot, base_objects=inherited / "objects",
+                                     expected_inherited_source_sha=inherited_source)
+            if merged.sccache_version != "sccache 0.18.0":
+                raise RuntimeError("unsupported sccache version")
+            if merged.base_image is not None and merged.base_image != labels["world.genarrative.ci.rust-cache-base"]:
+                raise RuntimeError("artifact base differs from its actual source image")
+            rustc = self.docker("run", "--rm", "--network", "none", "--read-only", base, "rustc", "-vV")
+            if rustc.strip() != merged.rustc.strip():
+                raise RuntimeError("artifact toolchain differs from target base image")
+            if merged.workspace != "/workspace/" + self.config["repository"]:
+                raise RuntimeError("artifact workspace differs from CI checkout")
+            shutil.copyfile(inherited / "sccache", snapshot / "sccache")
+            (snapshot / "sccache").chmod(0o755)
+            (snapshot / "base-image.txt").write_text(base + "\n")
+            (work / "Dockerfile").write_text(
+                f"FROM {base}\nCOPY snapshot/ /opt/genarrative-ci/rust-cache/\n"
+                f'LABEL world.genarrative.ci.rust-cache-source="{sha}"\n'
+                f'LABEL world.genarrative.ci.rust-cache-base="{base}"\n')
+            (work / ".dockerignore").write_text("**\n!Dockerfile\n!snapshot/\n!snapshot/**\n")
+            with build_log.open("a") as out:
+                self.docker("build", "--pull=false", "--tag", tag, str(work), output=out, timeout=1800)
+            self.build_command(build_log, "gitea-ci-job-image.sh", "verify", tag, env=env)
         image = self.image_info(tag)["Id"]
-        record = {"image": image, "tag": tag, "source": sha, "inputs": inputs,
-                  "base": base, "owned": True, "verified_run": None,
-                  "artifact": str(artifact)}
-        # 先登记再导出；进程中断后下次可以复用候选，不留下无归属成功镜像。
-        self.state["versions"].append(record)
+        self.state["versions"].append({**attempt, "image": image, "base": base,
+                                        "owned": True, "verified_run": None})
         self.state["candidate"] = image
         self.state["attempts"] = [row for row in self.state["attempts"] if row["source"] != sha]
         self.save()
-        log(f"candidate built source={sha} image={image}")
+        log(f"candidate assembled run={source['run_id']} source={sha} image={image}")
 
     def stage_candidate(self):
         candidate = self.version(self.state["candidate"])
@@ -309,10 +464,14 @@ class Maintenance:
         self.save()
 
     def idle(self):
-        runner = self.api.request(self.runner_api)
-        active = self.api.request("admin/actions/runs?status=in_progress&limit=1")
-        return (runner.get("busy") is False and active["total_count"] == 0
-                and not self.docker("ps", "-q", inner=True).strip())
+        gate = self.gate("status")
+        if type(gate.get("active_tasks")) is not int or gate["active_tasks"] < 0:
+            raise RuntimeError("gate must support durable task tracking before automatic switching")
+        # 已领取但尚未建容器、正在收尾上报的任务都由入口跟踪，不查管理员 API。
+        return (gate.get("uncertain") is False and gate.get("active_tasks") == 0
+                and not self.docker("ps", "-q", "--filter", "status=running",
+                                    "--filter", "status=created", "--filter", "status=restarting",
+                                    "--filter", "status=paused", inner=True).strip())
 
     def verify_current(self):
         current = self.version(self.state["current"])
@@ -402,12 +561,62 @@ class Maintenance:
                 del self.state["bases"][image]
                 self.save()
 
+    def cleanup_exports(self):
+        """Only our named artifacts; keep logs/runs and every unrelated artifact."""
+        protected_runs = {row["run_id"] for row in self.state.get("attempts", []) if row.get("run_id")}
+        for row in self.state["versions"]:
+            if row.get("run_id") and not row.get("staged"):
+                protected_runs.add(row["run_id"])
+            if not row.get("staged"):
+                continue
+            for item in row.get("exports", []):
+                if item.get("deleted"):
+                    continue
+                self.api.request(self.repo_api + f'/actions/artifacts/{item["id"]}', method="DELETE")
+                item["deleted"] = True
+                self.save()
+        cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=7)
+        # 先收集再删除，避免按页删除让下一页位置前移、漏掉旧产物。
+        artifacts = list(self.api.pages(self.repo_api + "/actions/artifacts", "artifacts"))
+        pattern = re.compile(re.escape(ARTIFACT_PREFIX) + "(" + "|".join(RUST_JOB_IDS.values())
+                             + r")-attempt-\d+\Z")
+        runs = {}
+        for item in artifacts:
+            run_id = (item.get("workflow_run") or {}).get("id")
+            if (not pattern.fullmatch(item["name"]) or not run_id or run_id in protected_runs
+                    or timestamp(item["created_at"]) >= cutoff):
+                continue
+            # Artifact.workflow_run 在 Gitea 1.26.4 中只有 id/repository_id/head_sha。
+            if run_id not in runs:
+                runs[run_id] = self.api.request(self.repo_api + f"/actions/runs/{run_id}")
+            run = runs[run_id]
+            if not self.master_run(run) or run.get("status") != "completed":
+                continue
+            self.api.request(self.repo_api + f'/actions/artifacts/{item["id"]}', method="DELETE")
+
+    def cleanup_pending_uploads(self):
+        # Gitea 1.26.4 的过期/DELETE API 不会清理未 finalized 的 V4 分块。
+        # 只处理本上传器命名的块，且 run 和文件本身均已过保留期限。
+        cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=7)
+        eligible = set()
+        protected = {row["run_id"] for row in self.state.get("attempts", []) if row.get("run_id")}
+        for run in self.api.pages(self.repo_api + "/actions/runs?branch=master&event=push", "workflow_runs"):
+            if (self.master_run(run) and run.get("status") == "completed"
+                    and run["id"] not in protected and run.get("completed_at")
+                    and timestamp(run["completed_at"]) < cutoff):
+                eligible.add(run["id"])
+        removed = cleanup_upload_chunks(Path(self.config["artifact_storage_dir"]), eligible, cutoff.timestamp())
+        if removed:
+            log(f"removed {removed} expired owned upload fragments")
+
     def tick(self, retry=False):
         if not self.recover_switch():
             return
         self.adopt_current()
-        sha, inputs = self.fetch_source()
+        self.fetch_source()
         self.recover_builds()
+        self.cleanup_exports()
+        self.cleanup_pending_uploads()
         if not self.verify_current():
             return
         self.cleanup()
@@ -415,35 +624,42 @@ class Maintenance:
             self.stage_candidate()
             self.activate()
             return
-        current = self.version(self.state["current"])
-        if "inputs" not in current and current.get("source"):
-            current["inputs"] = cache_inputs(command("git", "ls-tree", "-rz", current["source"], cwd=self.repo))
-            self.save()
-        if current.get("inputs") == inputs:
-            log(f"unchanged compilation inputs at master={sha}")
+        source = self.source_run()
+        if source is None:
+            log("waiting for a complete set of master CI cache exports")
             return
-        if not retry and self.state.get("failed_source") == sha:
-            log(f"previous build failed at {sha}; waiting for new master or --retry")
+        if not retry and self.state.get("failed_run") == source["run_id"]:
+            log(f'previous assembly failed at run={source["run_id"]}; waiting for new run or --retry')
             return
-        # 新的 CI 波峰不额外争抢预热资源，等下一次维护周期。
+        # 下载、合并和镜像装载也消耗宿主 IO；繁忙时留给 CI，下轮再收集。
         if not self.idle():
             log("CI active; defer refresh")
             return
         try:
-            self.build(sha, inputs)
+            self.build(source)
         except Exception:
-            self.state["failed_source"] = sha
+            self.state["failed_run"] = source["run_id"]
             self.save()
             raise
-        self.state.pop("failed_source", None)
+        self.state.pop("failed_run", None)
         self.save()
         self.stage_candidate()
+        self.cleanup_exports()
         self.activate()
 
     def recover_builds(self):
         # 宕机可能发生在 docker build 完成之后、登记 Image ID 之前。
         # 只接管预先登记的确定性 tag；恢复的候选仍须经过 stage 的 verify/load。
         for attempt in list(self.state.get("attempts", [])):
+            artifact = self.root / "artifacts" / attempt["source"]
+            if (not SHA.fullmatch(attempt["source"]) or Path(attempt["artifact"]).resolve() != artifact
+                    or artifact.is_symlink() or artifact.parent.is_symlink()):
+                raise RuntimeError("interrupted assembly directory is outside managed artifacts")
+            # 持有维护锁，只有此前中断的组装可能遗留这些私有工作目录。
+            for directory in artifact.glob("assemble-*"):
+                if directory.is_symlink() or not directory.is_dir():
+                    raise RuntimeError("unexpected interrupted assembly entry")
+                shutil.rmtree(directory)
             for kind in ("base_tag", "tag"):
                 ids = set(self.docker("image", "ls", "--no-trunc", "--quiet", attempt[kind]).split())
                 if not ids:
@@ -508,16 +724,14 @@ class Maintenance:
         gate = self.gate("status")
         # .runner 会因 label 更新而回写；mtime 不能证明内存中的地址。
         # 要求入口实际见到本次容器启动后、来自它的 FetchTask。
-        if gate.get("last_fetch_peer") not in addresses or gate.get("last_fetch_at", 0) < started:
+        if gate.get("last_fetch_peer") not in addresses or (gate.get("last_fetch_at") or 0) < started:
             raise RuntimeError("gate has not observed FetchTask from this runner startup")
+        if type(gate.get("active_tasks")) is not int or gate["active_tasks"] < 0:
+            raise RuntimeError("gate must support durable task tracking before automatic switching")
 
     def activate(self):
         self.check_gate_route()
         candidate = self.version(self.state["candidate"])
-        runner = self.api.request(self.runner_api)
-        if runner.get("disabled") is not False:
-            log("runner disabled by operator; defer switch")
-            return False
         config = self.read_config()
         gate = self.gate("status")
         if gate.get("paused") is not False and not self.state.get("pause_owned"):
@@ -608,7 +822,7 @@ def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--config", required=True)
     parser.add_argument("--apply", action="store_true", help="执行维护；默认只检查连接和配置")
-    parser.add_argument("--retry", action="store_true", help="重试同一 master 上失败的构建")
+    parser.add_argument("--retry", action="store_true", help="重试同一 master run 上失败的缓存组装")
     parser.add_argument("--resume", action="store_true", help="仅恢复本维护器暂停的领取；用于 ExecStopPost")
     args = parser.parse_args()
     if os.environ.get("CI") == "true":
@@ -617,14 +831,15 @@ def main():
     config = json.loads(Path(args.config).read_text())
     maintenance = Maintenance(config)
     if not args.apply and not args.resume:
-        runner = maintenance.api.request(maintenance.runner_api)
+        maintenance.api.request(maintenance.repo_api + "/actions/artifacts?limit=1")
         image = configured_image(maintenance.read_config())
         head = command("git", "ls-remote", config["clone_url"], "refs/heads/master").split()[0]
         maintenance.check_gate_route()
+        cleanup_upload_chunks(Path(config["artifact_storage_dir"]), set(), time.time())
         gate = maintenance.gate("status")
         if gate.get("uncertain"):
             raise RuntimeError("runner gate needs manual inspection")
-        log(f'check master={head} image={image} runner_disabled={runner["disabled"]} busy={runner["busy"]}')
+        log(f'check master={head} image={image} active_tasks={gate.get("active_tasks")}')
         return
     import fcntl  # Linux 宿主；纯逻辑测试仍可在 Windows 上导入。
     maintenance.root.mkdir(parents=True, exist_ok=True)

scripts/project-ci-workflow.test.ts

@@ -142,6 +142,38 @@ function backendStepIndex(stepName: string) {
 }
 
 describe('project CI workflow', () => {
+  it('publishes cache deltas only after Rust reporting on non-cancelled master pushes', () => {
+    const producers = [
+      'ai-game-creator-shell-rust-lane-1',
+      'ai-game-creator-shell-rust-lane-2',
+      'ai-game-creator-shell-rust-smoke',
+      'ai-game-creator-shell-rust-crates',
+      'backend-tests',
+      'native-shell-tests',
+    ];
+    for (const job of jobNames) {
+      if (!producers.includes(job)) {
+        expect(jobSection(job)).not.toContain('export-gitea-rust-cache.py');
+        continue;
+      }
+      const publish = stepSection(job, 'Publish master Rust cache artifact');
+      expect(publish).toContain(
+        "if: ${{ !cancelled() && github.event_name == 'push' && github.ref == 'refs/heads/master' }}",
+      );
+      expect(publish).toContain('GENARRATIVE_GITEA_TOKEN: ${{ github.token }}');
+      expect(publish).toContain('continue-on-error: true');
+      expect(publish).toContain(
+        'run: python3 scripts/export-gitea-rust-cache.py',
+      );
+      const section = jobSection(job);
+      expect(
+        section.indexOf('Publish master Rust cache artifact'),
+      ).toBeGreaterThan(
+        section.indexOf('run: bash scripts/ci-rust-cache.sh report'),
+      );
+    }
+  });
+
   it('uses isolated compilation caching for every Rust test job', () => {
     const firstRustSteps = {
       'ai-game-creator-shell-rust-lane-1':

scripts/test_gitea_cache_export.py

@@ -0,0 +1,205 @@
+"""master 产物的增量归档与 Gitea v4 上传协议回归。"""
+
+import hashlib
+import base64
+from contextlib import redirect_stdout
+from datetime import datetime, timezone
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import importlib.util
+import io
+import json
+import os
+from pathlib import Path
+import tarfile
+import tempfile
+import threading
+import unittest
+from unittest.mock import patch
+import urllib.parse
+import zipfile
+
+from gitea_cache_upload_cleanup import decode_owner
+
+
+spec = importlib.util.spec_from_file_location("cache_export", Path(__file__).with_name("export-gitea-rust-cache.py"))
+exporter = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(exporter)
+
+
+class ExportTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        (self.root / "objects").mkdir()
+
+    def object(self, key, value, mtime):
+        path = self.root / "objects" / key[0] / key[1] / key
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(value)
+        os.utime(path, ns=(mtime * 10**9, mtime * 10**9))
+        return path
+
+    def unpack(self, baseline, limit=1024):
+        destination = self.root / "snapshot.zip"
+        result = exporter.pack_snapshot(self.root, destination, {"job": "fixture"}, baseline, limit)
+        with zipfile.ZipFile(destination) as bundle:
+            self.assertEqual(bundle.namelist(), ["snapshot.tar"])
+            with tarfile.open(fileobj=io.BytesIO(bundle.read("snapshot.tar"))) as archive:
+                files = {member.name: archive.extractfile(member).read() for member in archive}
+        return result, json.loads(files.pop("manifest.json")), files
+
+    def test_delta_omits_hit_bytes_preserves_recency_and_exports_new_keys(self):
+        inherited = self.object("a" * 64, b"old", 10)
+        baseline_path = self.root / "baseline.json"
+        exporter.save_baseline(self.root, baseline_path)
+        baseline = json.loads(baseline_path.read_text())
+        # sccache 对命中对象更新 mtime，不能因此重复上传已有对象。
+        os.utime(inherited, ns=(20 * 10**9, 20 * 10**9))
+        self.object("b" * 64, b"new", 30)
+        result, manifest, files = self.unpack(baseline)
+        key = "objects/b/b/" + "b" * 64
+        self.assertEqual(result, (1, 3))
+        self.assertEqual(files, {key: b"new"})
+        self.assertEqual(manifest["mode"], "delta")
+        self.assertEqual(manifest["touched"], [{"path": "objects/a/a/" + "a" * 64, "mtime_ns": 20 * 10**9}])
+        self.assertEqual(manifest["objects"][0]["sha256"], hashlib.sha256(b"new").hexdigest())
+
+    def test_empty_delta_is_valid_and_changed_size_is_exported(self):
+        path = self.object("a" * 64, b"old", 10)
+        baseline = {"a/a/" + "a" * 64: {"size": 3, "mtime_ns": 10 * 10**9}}
+        self.assertEqual(self.unpack(baseline)[0], (0, 0))
+        path.write_bytes(b"replacement")
+        self.assertEqual(self.unpack(baseline)[0], (1, 11))
+
+    def test_capacity_prefers_newest_objects_and_excludes_unrelated_files(self):
+        self.object("a" * 64, b"old", 10)
+        self.object("b" * 64, b"new", 20)
+        (self.root / "objects" / "credentials").write_bytes(b"secret")
+        (self.root / "sccache").write_bytes(b"binary")
+        result, _, files = self.unpack({}, limit=3)
+        self.assertEqual(result, (1, 3))
+        self.assertEqual(list(files.values()), [b"new"])
+
+    @unittest.skipUnless(os.name == "posix", "symlink fixture requires POSIX")
+    def test_symlink_objects_and_subdirectories_are_not_exported(self):
+        outside = self.root / "secret"
+        outside.write_bytes(b"secret")
+        linked = self.object("a" * 64, b"old", 10)
+        linked.unlink()
+        linked.symlink_to(outside)
+        (self.root / "objects" / "b").symlink_to(self.root, target_is_directory=True)
+        self.assertEqual(self.unpack({})[0], (0, 0))
+
+    def test_non_master_and_unclean_shutdown_do_not_scan_or_upload(self):
+        with patch.object(exporter, "scan_objects", side_effect=AssertionError("must not scan")), \
+                patch.object(exporter, "upload_snapshot", side_effect=AssertionError("must not upload")):
+            exporter.export({"GITHUB_EVENT_NAME": "pull_request", "GITHUB_REF": "refs/heads/master"})
+            exporter.export({"GITHUB_EVENT_NAME": "push", "GITHUB_REF": "refs/heads/feature"})
+            exporter.export({"GITHUB_EVENT_NAME": "push", "GITHUB_REF": "refs/heads/master"})
+
+    def test_internal_repository_route_bypasses_rpc_only_gateway(self):
+        env = {
+            "GITHUB_REPOSITORY": "owner/repo",
+            "GITHUB_SERVER_URL": "http://fetch-gate:8080",
+            "GENARRATIVE_GITEA_REPOSITORY_URL": "http://gitea:3000/sub/owner/repo.git",
+        }
+        self.assertEqual(exporter.artifact_base_url(env), "http://gitea:3000/sub")
+        env["GENARRATIVE_GITEA_REPOSITORY_URL"] = "http://token@gitea:3000/owner/repo.git"
+        with self.assertRaises(ValueError):
+            exporter.artifact_base_url(env)
+
+    def test_native_v4_chunks_finalize_with_verified_size_and_hash(self):
+        calls = []
+
+        class Handler(BaseHTTPRequestHandler):
+            def log_message(self, *args):
+                pass
+
+            def do_POST(self):
+                body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
+                calls.append((self.command, self.path, dict(self.headers), body))
+                payload = {"ok": True}
+                if self.path.endswith("CreateArtifact"):
+                    payload["signedUploadUrl"] = "https://external.invalid/twirp/github.actions.results.api.v1.ArtifactService/UploadArtifact?sig=private&taskID=4&artifactID=8"
+                else:
+                    payload["artifactId"] = "8"
+                self.send_response(200)
+                self.end_headers()
+                self.wfile.write(json.dumps(payload).encode())
+
+            def do_PUT(self):
+                body = self.rfile.read(int(self.headers["Content-Length"]))
+                calls.append((self.command, self.path, dict(self.headers), body))
+                self.send_response(201)
+                self.end_headers()
+                self.wfile.write(b'"created"')
+
+        server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            path = self.root / "test.zip"
+            path.write_bytes(b"123456789")
+            with patch.object(exporter, "CHUNK_BYTES", 4):
+                exporter.upload_snapshot(path, "rust-cache-v1-backend-tests-attempt-1", 123, f"http://127.0.0.1:{server.server_port}", "ephemeral")
+        finally:
+            server.shutdown()
+            server.server_close()
+            thread.join()
+        self.assertEqual(len(calls), 6)
+        self.assertEqual(calls[0][3]["workflowRunBackendId"], "123")
+        remaining = datetime.fromisoformat(calls[0][3]["expiresAt"]) - datetime.now(timezone.utc)
+        self.assertGreater(remaining.total_seconds(), 7 * 24 * 3600)
+        self.assertLessEqual(remaining.total_seconds(), 8 * 24 * 3600)
+        blocks = calls[1:4]
+        self.assertEqual(b"".join(call[3] for call in blocks), b"123456789")
+        self.assertEqual(len({urllib.parse.parse_qs(urllib.parse.urlsplit(call[1]).query)["blockid"][0] for call in blocks}), 3)
+        for call in blocks:
+            block_id = urllib.parse.parse_qs(urllib.parse.urlsplit(call[1]).query)["blockid"][0]
+            gitea_filename_suffix = base64.urlsafe_b64encode(block_id.encode()).decode()
+            self.assertEqual(decode_owner(gitea_filename_suffix), ("backend-tests", 1))
+        self.assertTrue(all("Authorization" not in call[2] for call in calls[1:5]))
+        self.assertEqual(calls[-1][2]["Authorization"], "Bearer ephemeral")
+        self.assertEqual(calls[-1][3]["size"], "9")
+        self.assertEqual(calls[-1][3]["hash"], "sha256:" + hashlib.sha256(b"123456789").hexdigest())
+
+    def test_upload_failure_never_finalizes_an_incomplete_artifact(self):
+        path = self.root / "test.zip"
+        path.write_bytes(b"data")
+        with patch.object(exporter, "request", side_effect=[
+            {"ok": True, "signedUploadUrl": "http://gitea/twirp/github.actions.results.api.v1.ArtifactService/UploadArtifact?sig=x"},
+            RuntimeError("upload failed"),
+        ]) as request:
+            with self.assertRaisesRegex(RuntimeError, "upload failed"):
+                exporter.upload_snapshot(path, "rust-cache-v1-backend-tests-attempt-1", 123, "http://gitea", "ephemeral")
+            self.assertEqual(request.call_count, 2)
+
+    def test_failed_optional_export_never_emits_host_completion_marker(self):
+        (self.root / "export-baseline.json").write_text("{}")
+        (self.root / "rustc.txt").write_text("rustc fixture\n")
+        (self.root / "workspace.txt").write_text(str(Path.cwd().resolve()))
+        (self.root / "source-commit.txt").write_text("a" * 40)
+        self.object("b" * 64, b"new", 20)
+        env = {
+            "GITHUB_EVENT_NAME": "push", "GITHUB_REF": "refs/heads/master",
+            "GENARRATIVE_CI_RUST_CACHE_EXPORT_READY": "1",
+            "GITHUB_JOB": "backend-tests", "GITHUB_SHA": "b" * 40,
+            "GITHUB_RUN_ID": "123", "GITHUB_RUN_ATTEMPT": "1",
+            "GENARRATIVE_GITEA_TOKEN": "ephemeral",
+            "GITHUB_REPOSITORY": "owner/repo",
+            "GENARRATIVE_GITEA_REPOSITORY_URL": "http://gitea/owner/repo.git",
+            "GENARRATIVE_CI_RUST_CACHE_ROOT": str(self.root),
+        }
+        output = io.StringIO()
+        with patch.object(exporter.subprocess, "check_output", side_effect=["rustc fixture\n", "sccache 0.18.0\n"]), \
+                patch.object(exporter, "upload_snapshot", side_effect=RuntimeError("upload failed")), \
+                redirect_stdout(output):
+            with self.assertRaisesRegex(RuntimeError, "upload failed"):
+                exporter.export(env)
+        self.assertNotIn("complete=true", output.getvalue())
+        self.assertFalse((self.root / "export-baseline.json").exists())
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/test_gitea_cache_gate.py

@@ -2,6 +2,7 @@
 
 import http.client
 import http.server
+import gzip
 import importlib.util
 import json
 import os
@@ -21,6 +22,29 @@ MODULE = importlib.util.module_from_spec(SPEC)
 SPEC.loader.exec_module(MODULE)
 FETCH = "/api/actions/runner.v1.RunnerService/FetchTask"
 UPDATE = "/api/actions/runner.v1.RunnerService/UpdateTask"
+LOG = "/api/actions/runner.v1.RunnerService/UpdateLog"
+PING = "/api/actions/ping.v1.PingService/Ping"
+
+
+def varint(value):
+    data = bytearray()
+    while value > 127:
+        data.append((value & 127) | 128)
+        value >>= 7
+    data.append(value)
+    return bytes(data)
+
+
+def field(number, value):
+    # actions-proto-go v0.4.1: task/state ID=1; result=2; log task_id=1,
+    # index=2, rows=3, no_more=4; log response ack_index=1.
+    if isinstance(value, bytes):
+        return varint(number * 8 + 2) + varint(len(value)) + value
+    return varint(number * 8) + varint(value)
+
+
+def state(task_id, result=0):
+    return field(1, field(1, task_id) + field(2, result))
 
 
 def core_state(state):
@@ -50,8 +74,17 @@ class Upstream(http.server.BaseHTTPRequestHandler):
             self.wfile.write(b"incomplete")
             self.close_connection = True
             return
-        body = b'{"task":null}'
-        self.send_response(200)
+        body = self.server.responses.get(self.path, b"")
+        if self.path == UPDATE and self.path not in self.server.responses:
+            body = state(*map(int, MODULE.task_state(MODULE.protobuf_fields(data))))
+        if self.path == LOG and self.path not in self.server.responses:
+            fields = MODULE.protobuf_fields(data)
+            body = field(1, MODULE.single(fields, 2, 0) + len(fields.get(3, [])))
+        self.send_response(self.server.statuses.get(self.path, 200))
+        self.send_header("Content-Type", self.server.content_type)
+        if self.server.compressed:
+            body = gzip.compress(body)
+            self.send_header("Content-Encoding", "gzip")
         self.send_header("Content-Length", str(len(body)))
         self.end_headers()
         self.wfile.write(body)
@@ -66,6 +99,10 @@ class GateTests(unittest.TestCase):
         self.upstream.entered = threading.Event()
         self.upstream.release = threading.Event()
         self.upstream.truncated = False
+        self.upstream.responses = {}
+        self.upstream.statuses = {}
+        self.upstream.content_type = "application/proto"
+        self.upstream.compressed = False
         self.proxy = MODULE.create_proxy(("127.0.0.1", 0),
             f"http://127.0.0.1:{self.upstream.server_port}", self.gate)
         self.servers = [self.upstream, self.proxy]
@@ -79,13 +116,15 @@ class GateTests(unittest.TestCase):
             server.server_close()
         self.temp.cleanup()
 
-    def request(self, path, body=b"{}", chunked=False):
+    def request(self, path, body=b"", chunked=False):
         connection = http.client.HTTPConnection("127.0.0.1", self.proxy.server_port, timeout=5)
         try:
             if chunked:
-                connection.request("POST", path, body=[body], encode_chunked=True)
+                connection.request("POST", path, body=[body], encode_chunked=True,
+                                   headers={"Content-Type": "application/proto"})
             else:
-                connection.request("POST", path, body=body)
+                connection.request("POST", path, body=body,
+                                   headers={"Content-Type": "application/proto"})
             response = connection.getresponse()
             result = response.status, response.read()
             return result
@@ -112,7 +151,7 @@ class GateTests(unittest.TestCase):
         self.assertEqual(core_state(self.gate.control("pause")),
             {"paused": True, "inflight": 1, "uncertain": False})
         self.assertEqual(self.request(FETCH)[0], 503)
-        self.assertEqual(self.request(UPDATE)[0], 200)
+        self.assertEqual(self.request(PING)[0], 200)
         self.assertEqual(sum(path == FETCH for path, _ in self.upstream.requests), 1)
         self.upstream.release.set()
         thread.join(3)
@@ -159,8 +198,8 @@ class GateTests(unittest.TestCase):
         self.assertFalse(MODULE.Gate(self.temp.name).control("status")["paused"])
 
     def test_chunked_body_is_decoded_and_non_rpc_path_rejected(self):
-        self.assertEqual(self.request(UPDATE, b'{"state":"running"}', chunked=True)[0], 200)
-        self.assertEqual(self.upstream.requests[-1], (UPDATE, b'{"state":"running"}'))
+        self.assertEqual(self.request(PING, b"ping", chunked=True)[0], 200)
+        self.assertEqual(self.upstream.requests[-1], (PING, b"ping"))
         self.assertEqual(self.request("/api/v1/repos")[0], 404)
         self.assertEqual(self.request("/api/actions/../v1/repos")[0], 404)
 
@@ -200,12 +239,109 @@ class GateTests(unittest.TestCase):
         self.assertEqual(state["last_fetch_peer"], "127.0.0.1")
         self.assertGreaterEqual(state["last_fetch_at"], before)
         self.assertEqual(self.upstream.requests, [])
-        self.assertEqual(self.request(UPDATE)[0], 200)
+        self.assertEqual(self.request(PING)[0], 200)
         self.assertEqual(self.gate.control("status")["last_fetch_at"], state["last_fetch_at"])
         restarted = MODULE.Gate(self.temp.name).control("status")
         self.assertIsNone(restarted["last_fetch_peer"])
         self.assertIsNone(restarted["last_fetch_at"])
 
+    def fetch_task(self, task_id=42):
+        self.upstream.responses[FETCH] = field(1, field(1, task_id)) + field(2, 10)
+        self.upstream.release.set()
+        self.assertEqual(self.request(FETCH)[0], 200)
+        self.assertEqual(self.gate.control("status")["task_ids"], [str(task_id)])
+
+    def finalize_log(self, task_id=42):
+        self.assertEqual(self.request(LOG, field(1, task_id) + field(2, 3)
+                                      + field(3, b"log row") + field(4, 1))[0], 200)
+        self.wait_for(lambda: self.gate.tasks.get(str(task_id)) is True)
+
+    def test_assigned_task_survives_pause_until_runner_finished_reporting(self):
+        self.fetch_task()
+        self.assertEqual(self.gate.control("pause")["active_tasks"], 1)
+        self.assertEqual(self.request(FETCH)[0], 503)
+        self.assertEqual(self.request(UPDATE, state(42, 1))[0], 200)
+        self.assertEqual(self.gate.control("status")["active_tasks"], 1)
+        self.finalize_log()
+        self.assertEqual(self.gate.control("status")["active_tasks"], 1)
+        self.assertEqual(self.request(UPDATE, state(42, 1))[0], 200)
+        self.wait_for(lambda: self.gate.control("status")["active_tasks"] == 0)
+        self.assertFalse(self.gate.control("status")["uncertain"])
+        # 终态响应在客户端超时后重试，不应再次阻断后续任务领取。
+        self.request(UPDATE, state(42, 1))
+        self.assertFalse(self.gate.control("status")["uncertain"])
+
+    def test_compressed_fetch_and_task_ledger_survive_restart(self):
+        self.upstream.compressed = True
+        self.fetch_task()
+        restored = MODULE.Gate(self.temp.name)
+        self.assertEqual(restored.control("status")["task_ids"], ["42"])
+        self.assertFalse(restored.control("status")["uncertain"])
+        self.proxy.gate = self.gate = restored
+        self.finalize_log()
+        self.request(UPDATE, state(42, 2))
+        self.wait_for(lambda: self.gate.control("status")["active_tasks"] == 0)
+        self.assertEqual(MODULE.Gate(self.temp.name).control("status")["active_tasks"], 0)
+
+    def test_server_cancellation_does_not_retire_executing_task(self):
+        self.fetch_task()
+        self.upstream.responses[UPDATE] = state(42, 3)
+        self.request(UPDATE, state(42))
+        self.assertEqual(self.gate.control("status")["active_tasks"], 1)
+        self.finalize_log()
+        self.request(UPDATE, state(42))
+        self.assertEqual(self.gate.control("status")["active_tasks"], 1)
+        self.request(UPDATE, state(42, 3))
+        self.wait_for(lambda: self.gate.control("status")["active_tasks"] == 0)
+
+    def test_partial_final_log_acknowledgement_keeps_task_active(self):
+        self.fetch_task()
+        self.upstream.responses[LOG] = field(1, 3)
+        self.request(LOG, field(1, 42) + field(2, 3) + field(3, b"row") + field(4, 1))
+        self.request(UPDATE, state(42, 1))
+        self.assertEqual(self.gate.control("status")["active_tasks"], 1)
+        del self.upstream.responses[LOG]
+        self.finalize_log()
+        self.request(UPDATE, state(42, 1))
+        self.wait_for(lambda: self.gate.control("status")["active_tasks"] == 0)
+
+    def test_final_state_waits_for_success_and_output_acknowledgement(self):
+        self.fetch_task()
+        self.finalize_log()
+        self.upstream.statuses[UPDATE] = 500
+        self.request(UPDATE, state(42, 1))
+        self.assertEqual(self.gate.control("status")["active_tasks"], 1)
+        self.upstream.statuses[UPDATE] = 200
+        output = field(2, field(1, b"key") + field(2, b"value"))
+        self.request(UPDATE, state(42, 1) + output)
+        self.assertEqual(self.gate.control("status")["active_tasks"], 1)
+        self.upstream.responses[UPDATE] = state(42, 1) + field(2, b"key")
+        self.request(UPDATE, state(42, 1) + output)
+        self.wait_for(lambda: self.gate.control("status")["active_tasks"] == 0)
+
+    def test_unknown_or_invalid_fetch_result_prevents_idle(self):
+        self.upstream.release.set()
+        self.upstream.responses[FETCH] = b"not protobuf"
+        self.assertEqual(self.request(FETCH)[0], 200)
+        self.assertTrue(self.gate.control("status")["uncertain"])
+        self.assertTrue(MODULE.Gate(self.temp.name).control("status")["uncertain"])
+
+    def test_task_report_without_observed_assignment_requires_idle_bootstrap(self):
+        self.assertEqual(self.request(UPDATE, state(42))[0], 200)
+        self.wait_for(lambda: self.gate.control("status")["uncertain"])
+        self.assertIn("error", self.gate.control("resume"))
+
+    def test_disconnected_fetch_client_still_leaves_assigned_task_busy(self):
+        self.upstream.responses[FETCH] = field(1, field(1, 42))
+        client = socket.create_connection(self.proxy.server_address, timeout=3)
+        client.sendall(f"POST {FETCH} HTTP/1.1\r\nHost: localhost\r\nContent-Length: 0\r\n\r\n".encode())
+        self.assertTrue(self.upstream.entered.wait(3))
+        client.close()
+        self.gate.control("pause")
+        self.upstream.release.set()
+        self.wait_for(lambda: self.gate.control("status")["inflight"] == 0)
+        self.assertEqual(self.gate.control("status")["task_ids"], ["42"])
+
 
 if __name__ == "__main__":
     unittest.main()

scripts/test_gitea_cache_maintenance.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import importlib.util
+import io
 import json
 from pathlib import Path
 import tempfile
@@ -28,14 +29,11 @@ def runner_config(image: str) -> str:
 
 
 class FakeApi:
-    def __init__(self, disabled: bool = False):
-        self.disabled = disabled
+    def __init__(self):
         self.requests: list[str] = []
 
     def request(self, path, body=None, raw=False):
         self.requests.append(path)
-        if path.startswith("admin/actions/runners/"):
-            return {"disabled": self.disabled, "busy": False}
         raise AssertionError(f"unexpected API request: {path}")
 
 
@@ -51,7 +49,6 @@ class GiteaCacheMaintenanceTest(unittest.TestCase):
             "token_file": str(self.token),
             "repository": "team/project",
             "repository_url": "http://gitea:3000/team/project.git",
-            "runner_api_path": "admin/actions/runners/1",
             "runner_container": "gitea-runner",
             "gate_socket": str(self.root / "gate.sock"),
             "gate_url": "https://gitea.example.test",
@@ -168,7 +165,8 @@ class GiteaCacheMaintenanceTest(unittest.TestCase):
             return json.dumps([{"State": {"StartedAt": started}, "NetworkSettings": {
                 "Networks": {"internal": {"IPAddress": "10.0.0.2"}}}}])
         instance.docker = docker
-        proof = {"last_fetch_peer": "10.0.0.2", "last_fetch_at": maintenance_module.timestamp(started).timestamp() + 1}
+        proof = {"last_fetch_peer": "10.0.0.2", "last_fetch_at": maintenance_module.timestamp(started).timestamp() + 1,
+                 "active_tasks": 0}
         instance.gate = lambda action: proof
         instance.check_gate_route()
         proof["last_fetch_at"] -= 2
@@ -225,6 +223,186 @@ class GiteaCacheMaintenanceTest(unittest.TestCase):
         self.assertEqual(instance.state["candidate"], IMAGE)
         self.assertNotIn("pause_owned", instance.state)
 
+    def test_idle_requires_task_ledger_and_all_live_container_states(self):
+        instance = self.maintenance({"versions": [], "current": None})
+        state = {"uncertain": False, "active_tasks": 1}
+        instance.gate = lambda _: state
+        calls = []
+        instance.docker = lambda *args, **kw: calls.append((args, kw)) or ""
+        self.assertFalse(instance.idle())  # assigned task before its first container
+        self.assertEqual(calls, [])
+        state["active_tasks"] = 0
+        self.assertTrue(instance.idle())
+        self.assertIn("status=created", calls[-1][0])
+        self.assertTrue(calls[-1][1]["inner"])
+        instance.docker = lambda *_a, **_k: "created-container\n"
+        self.assertFalse(instance.idle())
+        del state["active_tasks"]
+        with self.assertRaisesRegex(RuntimeError, "durable task tracking"):
+            instance.idle()
+
+    def source_fixture(self):
+        old = self.record(OLD_IMAGE, source="b" * 40, owned=False, verified_run=1)
+        instance = self.maintenance({"versions": [old], "current": OLD_IMAGE})
+        run = {"id": 44, "status": "completed", "conclusion": "failure", "event": "push",
+               "path": "project-ci.yml@refs/heads/master", "head_branch": "master", "head_sha": SHA}
+        jobs = [{"id": i, "name": name, "status": "completed", "conclusion": "failure",
+                 "head_sha": SHA, "run_attempt": 1,
+                 "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]
+        class Api:
+            def pages(self, path, key):
+                return iter({"workflow_runs": [run], "jobs": jobs, "artifacts": artifacts}[key])
+            def request(self, path, raw=False):
+                job_id = int(path.split("/jobs/")[1].split("/")[0])
+                name = next(item["name"] for item in artifacts if item["id"] == job_id)
+                return f"worker image: {OLD_IMAGE}\n[rust-cache] artifact={name} objects=1 bytes=10 complete=true\n"
+        instance.api = Api()
+        return instance, run, jobs, artifacts
+
+    def select_source(self, instance):
+        with patch.object(maintenance_module.subprocess, "run", return_value=type("Result", (), {"returncode": 0})()), \
+             patch.object(maintenance_module, "command", lambda *args, **kw: "100644 blob " + args[-1] + "\tsrc.rs\0"):
+            return instance.source_run()
+
+    def test_complete_failed_master_can_supply_cache_but_pr_and_missing_groups_cannot(self):
+        instance, run, jobs, artifacts = self.source_fixture()
+        result = self.select_source(instance)
+        self.assertEqual(result["run_id"], 44)
+        self.assertEqual(len(result["exports"]), 6)
+        run["event"] = "pull_request"
+        self.assertIsNone(self.select_source(instance))
+        run["event"] = "push"
+        artifacts.pop()
+        self.assertIsNone(self.select_source(instance))
+
+    def test_stale_attempt_cancelled_job_or_failed_export_cannot_supply_cache(self):
+        instance, run, jobs, artifacts = self.source_fixture()
+        run["conclusion"] = "cancelled"
+        self.assertIsNone(self.select_source(instance))  # all six artifacts may already exist
+        run["conclusion"] = "failure"
+        jobs[0]["run_attempt"] = 2
+        self.assertIsNone(self.select_source(instance))
+        jobs[0]["run_attempt"] = 1
+        jobs[0]["conclusion"] = "cancelled"
+        self.assertIsNone(self.select_source(instance))
+        jobs[0]["conclusion"] = "success"
+        jobs[0]["steps"][0]["conclusion"] = "failure"
+        self.assertIsNone(self.select_source(instance))
+
+    def test_exports_using_different_images_are_not_combined(self):
+        instance, _, _, _ = self.source_fixture()
+        original = instance.api.request
+        instance.api.request = lambda path, raw=False: original(path, raw).replace(OLD_IMAGE, IMAGE) if '/jobs/1/' in path else original(path, raw)
+        self.assertIsNone(self.select_source(instance))
+
+    def test_completed_artifact_without_upload_completion_log_is_not_used(self):
+        instance, _, _, _ = self.source_fixture()
+        instance.api.request = lambda *_a, **_k: f"worker image: {OLD_IMAGE}\n"
+        self.assertIsNone(self.select_source(instance))
+
+    def test_assembly_uses_ci_objects_and_trusted_binary_without_warming_compiler(self):
+        instance, _, _, _ = self.source_fixture()
+        source = self.select_source(instance)
+        shell_calls, docker_calls = [], []
+        instance.build_command = lambda log, script, *args, **kw: shell_calls.append((script, args))
+        def info(image, inner=False):
+            return {"Id": IMAGE if image.startswith("genarrative/") else image, "Config": {"Labels": {
+                "world.genarrative.ci.rust-cache-source": "b" * 40,
+                "world.genarrative.ci.rust-cache-base": BASE_IMAGE,
+                "com.genarrative.ci.definition-sha256": "definition",
+            }}}
+        instance.image_info = info
+        def docker(*args, **kw):
+            docker_calls.append(args)
+            if args[0] == "create":
+                return "temporary-copy-container"
+            if args[0] == "cp":
+                root = Path(args[-1])
+                (root / "objects").mkdir()
+                (root / "sccache").write_bytes(b"trusted binary")
+            if args[-2:] == ("rustc", "-vV"):
+                return "rustc test\n"
+            return ""
+        instance.docker = docker
+        instance.api.download = lambda path, destination: destination.write_bytes(b"download")
+        def merge(inputs, output, **kwargs):
+            self.assertEqual(len(inputs), 6)
+            self.assertEqual(kwargs["expected_inherited_source_sha"], "b" * 40)
+            output.mkdir()
+            (output / "objects").mkdir()
+            return type("Merged", (), {"sccache_version": "sccache 0.18.0", "rustc": "rustc test\n",
+                "workspace": "/workspace/team/project", "base_image": BASE_IMAGE})()
+        with patch.object(maintenance_module, "command", return_value="definition\n"), \
+             patch.object(maintenance_module, "merge_snapshots", merge):
+            instance.build(source)
+        self.assertEqual(shell_calls, [("gitea-ci-job-image.sh", ("verify", "genarrative/gitea-project-ci:rust-cache-auto-" + SHA))])
+        self.assertIn(("create", OLD_IMAGE), docker_calls)
+        self.assertIn(("rm", "--volumes", "temporary-copy-container"), docker_calls)
+        self.assertEqual(instance.state["candidate"], IMAGE)
+        self.assertEqual(instance.version(IMAGE)["run_id"], 44)
+
+    def test_export_cleanup_only_deletes_collected_or_expired_owned_master_artifacts(self):
+        current = self.record(OLD_IMAGE, owned=False, verified_run=1)
+        current.update(staged=True, run_id=44, exports=[{"id": 1}])
+        instance = self.maintenance({"versions": [current], "current": OLD_IMAGE})
+        old_run = {"id": 9, "status": "completed", "path": "project-ci.yml@refs/heads/master",
+                   "event": "push", "head_branch": "master", "head_sha": SHA}
+        artifact = {"id": 2, "name": "rust-cache-v1-backend-tests-attempt-1",
+                    "workflow_run": {"id": 9}, "created_at": "2020-01-01T00:00:00Z"}
+        rows = [artifact, {**artifact, "id": 3, "name": "release-binary"},
+                {**artifact, "id": 4, "workflow_run": {"id": 10}},
+                {**artifact, "id": 5, "workflow_run": {"id": 11}}]
+        deleted = []
+        instance.api.pages = lambda *_: iter(rows)
+        def request(path, method=None):
+            if method == "DELETE":
+                return deleted.append((path, method))
+            return {9: old_run, 10: {**old_run, "event": "pull_request"},
+                    11: {**old_run, "status": "in_progress"}}[int(path.rsplit("/", 1)[1])]
+        instance.api.request = request
+        instance.cleanup_exports()
+        self.assertEqual(deleted, [(instance.repo_api + "/actions/artifacts/1", "DELETE"),
+                                   (instance.repo_api + "/actions/artifacts/2", "DELETE")])
+        self.assertTrue(instance.version(OLD_IMAGE)["exports"][0]["deleted"])
+
+    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"
+        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[0]}, None)
+                return io.BytesIO(b"archive")
+        with patch.object(maintenance_module.urllib.request, "build_opener", return_value=Opener()):
+            api.download("repos/team/project/actions/artifacts/1/zip", destination)
+            self.assertEqual(destination.read_bytes(), b"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")
+        self.assertFalse((self.root / "rejected.zip").exists())
+
+    def test_pending_chunk_cleanup_requires_old_terminal_master_run(self):
+        instance = self.maintenance({"versions": [], "current": None, "attempts": [{"run_id": 4}]})
+        instance.config["artifact_storage_dir"] = str(self.root / "storage")
+        run = {"id": 1, "status": "completed", "path": "project-ci.yml@refs/heads/master",
+               "event": "push", "head_branch": "master", "head_sha": SHA,
+               "completed_at": "2020-01-01T00:00:00Z"}
+        rows = [run, {**run, "id": 2, "event": "pull_request"},
+                {**run, "id": 3, "status": "in_progress"}, {**run, "id": 4},
+                {**run, "id": 5, "completed_at": "2999-01-01T00:00:00Z"}]
+        instance.api.pages = lambda *_: iter(rows)
+        with patch.object(maintenance_module, "cleanup_upload_chunks", return_value=0) as cleanup:
+            instance.cleanup_pending_uploads()
+        self.assertEqual(cleanup.call_args.args[1], {1})
+
     def test_uncertain_or_inflight_gate_never_restarts(self):
         instance, _, restarts = self.activation_maintenance(idle_values=[True])
         statuses = 0

scripts/test_gitea_cache_snapshot.py

@@ -0,0 +1,240 @@
+#!/usr/bin/env python3
+"""Tests for streamed Gitea sccache snapshot validation and merging."""
+
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+import io
+import json
+from pathlib import Path
+import os
+import sys
+import tarfile
+import tempfile
+import unittest
+import zipfile
+
+
+SCRIPT = Path(__file__).with_name("gitea_cache_snapshot.py")
+SPEC = importlib.util.spec_from_file_location("gitea_cache_snapshot", SCRIPT)
+assert SPEC and SPEC.loader
+snapshot = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = snapshot
+SPEC.loader.exec_module(snapshot)
+
+REPOSITORY = "team/project"
+SOURCE_SHA = "a" * 40
+INHERITED_SOURCE_SHA = "c" * 40
+BASE_IMAGE = "sha256:" + "b" * 64
+RUSTC = "rustc 1.90.0\nbinary: rustc\n"
+WORKSPACE = "/workspace/GenarrativeAI/Genarrative"
+SCCACHE_VERSION = "0.18.0"
+
+
+def object_path(character: str) -> str:
+    key = character * 64
+    return f"objects/{key[0]}/{key[1]}/{key}"
+
+
+class SnapshotMergeTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temporary_directory = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
+        self.root = Path(self.temporary_directory.name).resolve()
+        self.base = self.root / "base-objects"
+        self.base.mkdir()
+
+    def tearDown(self) -> None:
+        self.temporary_directory.cleanup()
+
+    def identity(self, job: str, run_id: int) -> object:
+        return snapshot.ArtifactIdentity(REPOSITORY, run_id, 1, job, SOURCE_SHA)
+
+    def archive(
+        self,
+        name: str,
+        job: str,
+        run_id: int,
+        objects: list[tuple[str, bytes, int]],
+        *,
+        prefix: bool = True,
+        manifest_update=None,
+        extra_member: tuple[str, bytes, str] | None = None,
+        touched: list[tuple[str, int]] | None = None,
+    ) -> object:
+        rows = [
+            {
+                "path": path,
+                "size": len(contents),
+                "sha256": hashlib.sha256(contents).hexdigest(),
+                "mtime_ns": mtime_ns,
+            }
+            for path, contents, mtime_ns in objects
+        ]
+        manifest = {
+            "schema": 1,
+            "repository": REPOSITORY,
+            "run_id": run_id,
+            "run_attempt": 1,
+            "job": job,
+            "source_sha": SOURCE_SHA,
+            "base_image": BASE_IMAGE,
+            "rustc": RUSTC,
+            "workspace": WORKSPACE,
+            "sccache_version": SCCACHE_VERSION,
+            "mode": "delta",
+            "inherited_source_sha": INHERITED_SOURCE_SHA,
+            "objects": rows,
+            "touched": [
+                {"path": path, "mtime_ns": mtime_ns}
+                for path, mtime_ns in (touched or [])
+            ],
+        }
+        if manifest_update is not None:
+            manifest_update(manifest)
+
+        tar_bytes = io.BytesIO()
+        with tarfile.open(fileobj=tar_bytes, mode="w") as bundle:
+            for path, contents, _ in objects:
+                info = tarfile.TarInfo(path)
+                info.size = len(contents)
+                bundle.addfile(info, io.BytesIO(contents))
+            if extra_member is not None:
+                path, contents, kind = extra_member
+                info = tarfile.TarInfo(path)
+                if kind == "symlink":
+                    info.type = tarfile.SYMTYPE
+                    info.linkname = "manifest.json"
+                    bundle.addfile(info)
+                else:
+                    info.size = len(contents)
+                    bundle.addfile(info, io.BytesIO(contents))
+            manifest_bytes = json.dumps(manifest, separators=(",", ":")).encode()
+            info = tarfile.TarInfo("manifest.json")
+            info.size = len(manifest_bytes)
+            bundle.addfile(info, io.BytesIO(manifest_bytes))
+
+        archive = self.root / f"{name}.zip"
+        member = f"{name}/snapshot.tar" if prefix else "snapshot.tar"
+        with zipfile.ZipFile(archive, "w", compression=zipfile.ZIP_STORED) as bundle:
+            bundle.writestr(member, tar_bytes.getvalue())
+        return snapshot.ArtifactInput(archive, self.identity(job, run_id))
+
+    def base_object(self, path: str, contents: bytes, mtime_ns: int) -> None:
+        target = self.base.joinpath(*Path(path).parts[1:])
+        target.parent.mkdir(parents=True, exist_ok=True)
+        target.write_bytes(contents)
+        os.utime(target, ns=(mtime_ns, mtime_ns))
+
+    def merge(self, inputs, output=None, maximum=1024 ** 3):
+        return snapshot.merge_snapshots(
+            inputs,
+            output or self.root / "merged",
+            base_objects=self.base,
+            expected_inherited_source_sha=INHERITED_SOURCE_SHA,
+            max_combined_bytes=maximum,
+        )
+
+    def test_merges_groups_deduplicates_newest_mtime_and_prunes_to_bound(self) -> None:
+        first_path = object_path("a")
+        second_path = object_path("b")
+        third_path = object_path("c")
+        inherited_path = object_path("d")
+        self.base_object(inherited_path, b"dd", 50)
+        inputs = [
+            self.archive("first", "Backend tests", 41, [
+                (first_path, b"aaaa", 100),
+                (second_path, b"bbbb", 200),
+            ]),
+            self.archive("second", "Native shell tests", 41, [
+                (first_path, b"aaaa", 300),
+                (third_path, b"ccc", 250),
+            ], prefix=False, touched=[(inherited_path, 400)]),
+        ]
+
+        output = self.root / "merged"
+        result = self.merge(inputs, output, maximum=9)
+
+        self.assertEqual((result.object_count, result.total_bytes), (3, 9))
+        self.assertEqual((output / inherited_path).read_bytes(), b"dd")
+        self.assertEqual((output / first_path).read_bytes(), b"aaaa")
+        self.assertEqual((output / third_path).read_bytes(), b"ccc")
+        self.assertFalse((output / second_path).exists())
+        self.assertEqual((output / "rustc.txt").read_text(), RUSTC)
+        self.assertEqual((output / "workspace.txt").read_text(), WORKSPACE + "\n")
+        self.assertEqual((output / "source-commit.txt").read_text(), SOURCE_SHA + "\n")
+        self.assertEqual((output / "base-image.txt").read_text(), BASE_IMAGE + "\n")
+        self.assertFalse((output / "sccache").exists())
+
+    def test_rejects_traversal_member_without_creating_output(self) -> None:
+        item = self.archive(
+            "traversal", "Backend tests", 42, [],
+            extra_member=("../outside", b"bad", "file"),
+        )
+        output = self.root / "merged"
+        with self.assertRaisesRegex(snapshot.SnapshotError, "invalid sccache object path"):
+            self.merge([item], output)
+        self.assertFalse(output.exists())
+        self.assertFalse((self.root / "outside").exists())
+
+    def test_rejects_checksum_mismatch(self) -> None:
+        path = object_path("d")
+
+        def corrupt(manifest):
+            manifest["objects"][0]["sha256"] = "0" * 64
+
+        item = self.archive(
+            "checksum", "Backend tests", 43, [(path, b"content", 1)],
+            manifest_update=corrupt,
+        )
+        with self.assertRaisesRegex(snapshot.SnapshotError, "checksums do not match"):
+            self.merge([item])
+
+    def test_rejects_manifest_identity_mismatch(self) -> None:
+        item = self.archive(
+            "identity", "Backend tests", 44, [],
+            manifest_update=lambda manifest: manifest.update(run_id=999),
+        )
+        with self.assertRaisesRegex(snapshot.SnapshotError, "run_id does not match"):
+            self.merge([item])
+
+    def test_rejects_conflicting_duplicate_key(self) -> None:
+        path = object_path("e")
+        inputs = [
+            self.archive("conflict-one", "Backend tests", 45, [(path, b"one", 1)]),
+            self.archive("conflict-two", "Native shell tests", 45, [(path, b"two", 2)]),
+        ]
+        with self.assertRaisesRegex(snapshot.SnapshotError, "conflicting content"):
+            self.merge(inputs)
+
+    def test_rejects_delta_that_conflicts_with_inherited_key(self) -> None:
+        path = object_path("f")
+        self.base_object(path, b"base", 1)
+        item = self.archive("base-conflict", "Backend tests", 46, [(path, b"evil", 2)])
+        with self.assertRaisesRegex(snapshot.SnapshotError, "conflicting content"):
+            self.merge([item])
+
+    def test_accepts_empty_delta(self) -> None:
+        item = self.archive("empty", "Backend tests", 47, [])
+        output = self.root / "merged"
+        result = self.merge([item], output)
+        self.assertEqual((result.object_count, result.total_bytes), (0, 0))
+        self.assertEqual(list((output / "objects").iterdir()), [])
+
+    def test_rejects_noncanonical_base_entry(self) -> None:
+        (self.base / "unexpected").write_bytes(b"not an object")
+        item = self.archive("base-path", "Backend tests", 48, [])
+        with self.assertRaisesRegex(snapshot.SnapshotError, "noncanonical entry"):
+            self.merge([item])
+
+    def test_rejects_tar_links(self) -> None:
+        item = self.archive(
+            "link", "Backend tests", 49, [],
+            extra_member=(object_path("9"), b"", "symlink"),
+        )
+        with self.assertRaisesRegex(snapshot.SnapshotError, "regular file"):
+            self.merge([item])
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/test_gitea_cache_upload_cleanup.py

@@ -0,0 +1,90 @@
+"""缓存上传残块清理只触碰已确认结束的过期自有文件。"""
+
+import base64
+import os
+from pathlib import Path
+import tempfile
+import unittest
+
+from gitea_cache_upload_cleanup import cleanup_upload_chunks, decode_owner
+
+
+class UploadCleanupTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name).resolve()
+
+    def file(self, name, run=12, age=10):
+        directory = self.root / "tmp-upload" / f"run-{run}-v4"
+        directory.mkdir(parents=True, exist_ok=True)
+        path = directory / name
+        path.write_bytes(b"block data")
+        os.utime(path, (age, age))
+        return path
+
+    def block(self, artifact=34, run=12, age=10, marker="genarrative-rust-cache-v1:backend-tests:1:00000000"):
+        encoded = base64.urlsafe_b64encode(base64.b64encode(marker.encode())).decode()
+        return self.file(f"block-{run}-{artifact}-10-{encoded}", run, age)
+
+    def test_old_owned_blocks_and_associated_blocklist_are_deleted(self):
+        block = self.block()
+        blocklist = self.file("12-34-blocklist")
+        self.assertEqual(cleanup_upload_chunks(self.root, {12}, 20), 2)
+        self.assertFalse(block.exists())
+        self.assertFalse(blocklist.exists())
+        self.assertTrue(block.parent.is_dir())
+
+    def test_other_runs_artifacts_and_unproven_blocklists_remain(self):
+        own = self.block()
+        other_run = self.block(run=13)
+        other_artifact = self.block(artifact=35, marker="unrelated-upload")
+        unproven = self.file("12-36-blocklist")
+        self.assertEqual(cleanup_upload_chunks(self.root, {12}, 20), 1)
+        self.assertFalse(own.exists())
+        self.assertTrue(all(path.exists() for path in (other_run, other_artifact, unproven)))
+
+    def test_young_or_unknown_member_protects_entire_artifact_group(self):
+        own = self.block()
+        young = self.file("12-34-blocklist", age=20)
+        other = self.block(artifact=35)
+        unknown = self.file("block-12-35-invalid-size-unknown")
+        self.assertEqual(cleanup_upload_chunks(self.root, {12}, 20), 0)
+        self.assertTrue(all(path.exists() for path in (own, young, other, unknown)))
+
+    def test_completely_unknown_file_protects_entire_run(self):
+        block = self.block()
+        self.file("unknown")
+        self.assertEqual(cleanup_upload_chunks(self.root, {12}, 20), 0)
+        self.assertTrue(block.exists())
+
+    @unittest.skipUnless(os.name == "posix", "symlink fixture requires POSIX")
+    def test_symlinks_never_traversed_or_deleted(self):
+        block = self.block()
+        outside = self.root / "secret"
+        outside.write_text("keep")
+        (block.parent / "12-34-blocklist").symlink_to(outside)
+        (self.root / "tmp-upload" / "run-13-v4").symlink_to(block.parent, target_is_directory=True)
+        self.assertEqual(cleanup_upload_chunks(self.root, {12, 13}, 20), 0)
+        self.assertTrue(block.exists())
+        self.assertEqual(outside.read_text(), "keep")
+        linked = self.root / "root-link"
+        linked.symlink_to(self.root, target_is_directory=True)
+        with self.assertRaises(ValueError):
+            cleanup_upload_chunks(linked, {12}, 20)
+
+    def test_root_missing_or_relative_is_an_error(self):
+        with self.assertRaises(FileNotFoundError):
+            cleanup_upload_chunks(self.root / "missing", {12}, 20)
+        with self.assertRaises(ValueError):
+            cleanup_upload_chunks(Path("relative"), {12}, 20)
+
+    def test_noncanonical_encoding_and_unknown_job_are_not_owned(self):
+        for marker in ("genarrative-rust-cache-v1:other-job:1:00000000", "genarrative-rust-cache-v1:backend-tests:01:00000000"):
+            encoded = base64.urlsafe_b64encode(base64.b64encode(marker.encode())).decode()
+            self.assertIsNone(decode_owner(encoded))
+        self.assertIsNone(decode_owner("not-base64"))
+
+
+if __name__ == "__main__":
+    unittest.main()