补齐 CI 缓存维护器请求与任务契约校验

要求 API 请求显式指定 HTTP 方法并更新全部调用点
增加 workflow 全部任务名称与缓存导出映射的契约测试
从 Rust 任务映射推导集合和数量，消除重复硬编码
同步开发流程中的维护约定

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

@@ -98,6 +98,8 @@ SpacetimeDB 任务统一先读取 `.codex/skills/genarrative-spacetimedb/SKILL.m
 
 Gitea Rust 缓存自动维护由宿主 `genarrative-ci-cache.timer` 收集同一 master push run 六个 Rust job 的原生 V4 缓存产物，不重复执行 Cargo 预热。只传本轮新 key，命中对象只传使用时间；宿主与真实来源镜像对象合并、去重、按新近使用时间裁剪到 4 GiB，从无对象缓存基础镜像重新组装。源 run 不要求全绿，但取消、缺组、旧 attempt、未完成上传或混用来源镜像不得采用。网关暂停新 FetchTask、在途领取结束、持久化任务账本清空且内层活动容器为空才切换，不打断运行中的 CI。首次接入/升级网关需空闲窗口；Token 只需普通仓库 `write:repository`，不查管理员 API。候选装载后清理已收集 artifact，遗留项保留 7 天；真实 master CI 验证后才清理旧镜像，保留当前、一个回滚版、基础镜像及容器引用。部署入口见 `deploy/container/README.md`，合并代码不等于服务启用。
 
+修改 Gitea workflow 的 job 显示名称、ID 或缓存导出组时，必须同步维护器的 `JOBS` / `RUST_JOB_IDS`；`test_gitea_cache_maintenance.py` 直接对照实际 workflow 检查全集和导出映射，避免自动刷新或镜像验收因名单漂移长期等待。维护器 `Api.request` 的 `method` 是必填关键字参数，GET 也必须显式指定，不根据 body 推断请求方法。
+
 AGC Rust 两条 lane、crates、smoke、Backend 和桌面壳测试使用镜像内可信 sccache 对象快照；Native shell release step 显式清空双 wrapper，前端/repository checks 不启用。仅首次人工 bootstrap 时，维护者通过 `scripts/build-gitea-rust-cache.sh` 从远端 master 在限额、无宿主挂载的临时容器中按实际 cwd/profile/目标预热全部测试组，仅编译、不执行测试/应用；后端 workspace 与 spacetime-module 保持独立，AGC 的三个 cwd 入口之间清理预热 target，防止 fresh 判断漏产缓存键。最终镜像只追加 sccache、对象和来源元数据，不包含源码或 target。容量上限 4 GiB，不替代宿主旧镜像/归档清理。PR 只写当前容器层、不回传，不开放 Docker API/发布权限；继续禁用 incremental。`ci-rust-cache.sh` 在快照缺失、工具链不符或 wrapper 探测失败时直接编译，并隔离远程缓存配置和 daemon。分片日志记录编译耗时，收尾输出命中统计；两个 lane 的测试和前置检查不同，耗时差不是严格 A/B。线上存在活跃 CI 时不得重启 runner 或切换标签；全组启用前须刷新完整快照并逐组验证，详见开发运维文档。
 
 `.gitea/workflows/project-ci.yml` 的客户端门禁拆成 lane 与功能 job，每个 job 只预热自己会构建的那几份依赖：`AI game creator shell Rust lane 1/2`、`lane 2/2` 各自预取一次 AGC 壳 manifest，并顺序运行两片 Rust bin 单测；`AI game creator shell Rust smoke` 同样只预取 AGC 壳 manifest（`agent-run` smoke 会用 `src-tauri/Cargo.toml` spawn `cargo run`），`AI game creator shell Rust crates` 预取 `server-rs/Cargo.toml` 与独立 crate，`Native shell tests` 预取桌面壳与 AGC 壳 manifest，`AI game creator shell web tests` 不触碰 Cargo，不预热。两条 Rust lane、smoke job 与 crates job 只用 cargo 与 node 内建模块，因此不执行 `npm ci`。两个被 `server-rs/Cargo.toml` 排除、且没有提交 `Cargo.lock` 的独立 crate（`agent-runtime-core`、`agent-runtime-orchestration`）只能在 `AI game creator shell Rust crates` 里用不带锁标志的 fetch。AGC 壳的 bin target 单测（约 2466 条）由 `apps/ai-game-creator-shell/scripts/run-rust-shell-test-shards.mjs` 编译后按 `--list` 名单分 4 片：每次分片调用用 `--shard-index=<i>` 只跑自己那片，片内保持 `--test-threads=1` 并使用独立 `TMPDIR`；两条 lane 之间并发，lane 内顺序运行两片，避免重复依赖预热和同一容器内多进程争抢。不要改回「一个 job 内多进程并行这几片」——同一容器里它们会争抢共享 `HOME`、target 目录与固定临时路径，实测比整套串行还慢。每个分片调用都会自校验「片并集等于全集且互斥」，因此改分片规则不会静默漏跑。Backend host workspace tests 使用 `cargo test --locked --workspace --exclude spacetime-module --no-fail-fast`，避免 `spacetime-module` 的 `spacetime-types` feature 统一污染普通领域 crate 的 host 测试；随后单独执行 `cargo test --locked -p spacetime-module --no-fail-fast`，由 `spacetime-module/src/active.rs` 在 host 测试构建期间提供仅测试期的 SpacetimeDB ABI 链接支持，使该 crate 的纯单元测试也纳入 Backend 门禁。`spacetime-module` 的 reducer / procedure 运行时行为仍必须通过真实 SpacetimeDB runtime/integration harness 验证，host 链接支持不得被当作运行时替身。Backend 另外执行 `cargo check --locked -p spacetime-module` 验证模块源码。AGC 壳检查还会运行 `platform-llm` 与 `shared-contracts` 的 server-rs workspace 测试，这些命令以及 AGC 壳测试必须带 `--locked`，避免在测试阶段重新解析 registry index；锁文件发生变化时应先更新受信任 CI 镜像缓存，再重跑门禁。

scripts/maintain-gitea-rust-cache.py

@@ -33,7 +33,6 @@ JOBS = {
     "Backend tests", "Native shell tests", "Frontend tests",
     "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",
@@ -42,6 +41,7 @@ RUST_JOB_IDS = {
     "Backend tests": "backend-tests",
     "Native shell tests": "native-shell-tests",
 }
+RUST_JOBS = set(RUST_JOB_IDS)
 ARTIFACT_PREFIX = "rust-cache-v1-"
 MAX_DOWNLOAD = 4 * 1024 ** 3 + 129 * 1024 ** 2
 EXPORT_STEP = "Publish master Rust cache artifact"
@@ -143,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, method=None):
+    def request(self, path, *, method, body=None, raw=False):
         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=method or ("GET" if body is None else "PATCH"),
+            method=method,
             headers={"Authorization": "token " + token, "Content-Type": "application/json"},
         )
         try:
@@ -199,7 +199,7 @@ class Api:
         separator = "&" if "?" in path else "?"
         page = 1
         while True:
-            response = self.request(f"{path}{separator}limit=50&page={page}")
+            response = self.request(f"{path}{separator}limit=50&page={page}", method="GET")
             items = response[key]
             yield from items
             if len(items) < 50:
@@ -328,7 +328,7 @@ class Maintenance:
                 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:
+            if len(rust_jobs) != len(RUST_JOB_IDS) 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
@@ -344,7 +344,7 @@ class Maintenance:
                            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)
+                content = self.api.request(self.repo_api + f'/actions/jobs/{job["id"]}/logs', method="GET", raw=True)
                 if not re.search(re.escape(f"[rust-cache] artifact={name}")
                                  + r" objects=\d+ bytes=\d+ complete=true", content):
                     break
@@ -354,7 +354,7 @@ class Maintenance:
                 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:
+            if len(selected) != len(RUST_JOB_IDS) or len(images) != 1:
                 continue
             source_image = images.pop()
             if source_image not in {row["image"] for row in self.state["versions"]}:
@@ -493,7 +493,7 @@ class Maintenance:
             if not fully_passed(run, jobs):
                 continue
             if all(verifies_image(job, self.api.request(
-                self.repo_api + f'/actions/jobs/{job["id"]}/logs', raw=True), current["image"],
+                self.repo_api + f'/actions/jobs/{job["id"]}/logs', method="GET", raw=True), current["image"],
                 require_cache=current.get("source") is not None,
             ) for job in jobs):
                 current["verified_run"] = run["id"]
@@ -588,7 +588,7 @@ class Maintenance:
                 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}")
+                runs[run_id] = self.api.request(self.repo_api + f"/actions/runs/{run_id}", method="GET")
             run = runs[run_id]
             if not self.master_run(run) or run.get("status") != "completed":
                 continue
@@ -831,7 +831,7 @@ def main():
     config = json.loads(Path(args.config).read_text())
     maintenance = Maintenance(config)
     if not args.apply and not args.resume:
-        maintenance.api.request(maintenance.repo_api + "/actions/artifacts?limit=1")
+        maintenance.api.request(maintenance.repo_api + "/actions/artifacts?limit=1", method="GET")
         image = configured_image(maintenance.read_config())
         head = command("git", "ls-remote", config["clone_url"], "refs/heads/master").split()[0]
         maintenance.check_gate_route()

scripts/test_gitea_cache_maintenance.py

@@ -7,6 +7,7 @@ import importlib.util
 import io
 import json
 from pathlib import Path
+import re
 import tempfile
 import unittest
 from unittest.mock import patch
@@ -32,7 +33,7 @@ class FakeApi:
     def __init__(self):
         self.requests: list[str] = []
 
-    def request(self, path, body=None, raw=False):
+    def request(self, path, *, method, body=None, raw=False):
         self.requests.append(path)
         raise AssertionError(f"unexpected API request: {path}")
 
@@ -58,6 +59,41 @@ class GiteaCacheMaintenanceTest(unittest.TestCase):
     def tearDown(self):
         self.temporary_directory.cleanup()
 
+    def test_api_requires_explicit_method_and_preserves_it_with_body(self):
+        api = maintenance_module.Api(self.config["api_url"], self.token)
+        with self.assertRaises(TypeError):
+            api.request("example", body={"value": 1})
+        with self.assertRaises(TypeError):
+            api.request("example")
+        with patch.object(maintenance_module.urllib.request, "build_opener") as build_opener:
+            for method in ("GET", "POST", "PUT", "PATCH", "DELETE"):
+                with self.subTest(method=method):
+                    build_opener.return_value.open.return_value = io.BytesIO(b'{}')
+                    body = {"value": 1} if method in {"POST", "PUT", "PATCH"} else None
+                    self.assertEqual(api.request("example", method=method, body=body), {})
+                    request = build_opener.return_value.open.call_args.args[0]
+                    self.assertEqual(request.get_method(), method)
+                    self.assertEqual(request.data, None if body is None else json.dumps(body).encode())
+
+    def test_workflow_jobs_and_cache_producers_match_maintenance_contract(self):
+        workflow = (SCRIPT.parent.parent / ".gitea/workflows/project-ci.yml").read_text(encoding="utf-8")
+        # 沿用 workflow 的显式 job/step 格式，枚举实际 job，避免另一份名单漏掉新增项。
+        sections = re.split(r"(?m)^  ([\w-]+):\s*$", workflow.split("\njobs:\n", 1)[1])
+        jobs, producers = {}, {}
+        for job_id, section in zip(sections[1::2], sections[2::2]):
+            names = re.findall(r"(?m)^    name: (.+)$", section)
+            self.assertEqual(len(names), 1, f"{job_id}: expected one explicit job name")
+            name = names[0].strip().strip("\"'")
+            self.assertNotIn(name, jobs, f"duplicate workflow job name: {name}")
+            jobs[name] = job_id
+            if "run: python3 scripts/export-gitea-rust-cache.py" in section:
+                self.assertIn(f"      - name: {maintenance_module.EXPORT_STEP}\n", section)
+                producers[name] = job_id
+        self.assertEqual(set(jobs), maintenance_module.JOBS,
+                         "workflow job names drifted from maintainer JOBS")
+        self.assertEqual(producers, maintenance_module.RUST_JOB_IDS,
+                         "workflow cache producer names/IDs drifted from maintainer RUST_JOB_IDS")
+
     def record(self, image, *, source=SHA, owned=True, verified_run=None, base=BASE_IMAGE):
         return {
             "image": image,
@@ -255,7 +291,7 @@ class GiteaCacheMaintenanceTest(unittest.TestCase):
         class Api:
             def pages(self, path, key):
                 return iter({"workflow_runs": [run], "jobs": jobs, "artifacts": artifacts}[key])
-            def request(self, path, raw=False):
+            def request(self, path, *, method, 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"
@@ -295,7 +331,7 @@ class GiteaCacheMaintenanceTest(unittest.TestCase):
     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)
+        instance.api.request = lambda path, **kwargs: original(path, **kwargs).replace(OLD_IMAGE, IMAGE) if '/jobs/1/' in path else original(path, **kwargs)
         self.assertIsNone(self.select_source(instance))
 
     def test_completed_artifact_without_upload_completion_log_is_not_used(self):
@@ -357,7 +393,7 @@ class GiteaCacheMaintenanceTest(unittest.TestCase):
                 {**artifact, "id": 5, "workflow_run": {"id": 11}}]
         deleted = []
         instance.api.pages = lambda *_: iter(rows)
-        def request(path, method=None):
+        def request(path, *, method):
             if method == "DELETE":
                 return deleted.append((path, method))
             return {9: old_run, 10: {**old_run, "event": "pull_request"},
@@ -554,7 +590,7 @@ class GiteaCacheMaintenanceTest(unittest.TestCase):
                     return iter(jobs)
                 raise AssertionError(path)
 
-            def request(self, path, body=None, raw=False):
+            def request(self, path, *, method, body=None, raw=False):
                 if not raw or "/actions/jobs/" not in path:
                     raise AssertionError(path)
                 return f"runner image: {OLD_IMAGE}\n"