fix: 提前失败路也清附件暂存（仓库解析失败/缺 key 不再残留 bundle）

2026-10-27 附件实弹：仓库解析失败的任务不走正常终局清理，
bundle 目录永久残留（首个失败任务的暂存目录须手动删）。

根因：run_one 里缺 key 门与仓库解析失败两路提前返回，够不
到正常终局后的暂存清理；只有走完 _run_attempt 的任务才删。

修法：把终局清理抽成 _cleanup_task_seed 唯一入口，三路（正
常终局、缺 key 门、仓库解析失败）都走它；取消仍走 server
侧删，互不冲突；删失败不影响任务结果本身。

测试：test_worker.py 新增 2 个（仓库解析失败、缺 key 时
bundle 目录都被删）；反向验证：撤掉 worker 修复时正好这
2 个变红，有修复时全绿。全量 deploy 19、host 62、cloud
610 全过。

cloud/tests/test_worker.py

@@ -491,6 +491,54 @@ class WorkerTestCase(unittest.TestCase):
         self.assertEqual(result["error"],
                          "Missing environment variable: XCM_API_KEY")
 
+    # -- 提前失败路的暂存清理 --------------------------------------------
+    # 2026-10-27 实弹：仓库解析失败的任务不走正常终局清理，
+    # bundle 目录永久残留。缺 key 门是同款提前返回，同测。
+    def _attach_bundle(self, task_id, bundle_id="bundle1"):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO task_attachments(task_id, bundle_id,"
+                " rel_path, size) VALUES(?, ?, ?, ?)",
+                (task_id, bundle_id, "a.txt", 3))
+            conn.commit()
+        finally:
+            conn.close()
+        bundle_dir = os.path.join(self.upload_dir, bundle_id)
+        os.makedirs(bundle_dir, exist_ok=True)
+        with open(os.path.join(bundle_dir, "a.txt"), "w") as f:
+            f.write("x")
+        return bundle_dir
+
+    def test_repo_failure_cleans_staging_bundle(self):
+        _insert_task(self.db_path, "A")
+        self.upload_dir = os.path.join(self.tmp.name, "uploads")
+        bundle_dir = self._attach_bundle("A")
+        worker = self._worker(
+            runner=self._recording_runner(_completed_runner),
+            upload_dir=self.upload_dir)
+        worker._project_repo_url = lambda pid: "https://example.invalid/r.git"
+
+        def _boom(cmd, timeout=None):
+            raise Exception("clone boom")
+
+        worker.repos_runner = _boom
+        result = worker.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("仓库解析失败", result["error"])
+        self.assertEqual(_status(self.db_path, "A"), "failed")
+        self.assertEqual(self.calls, [])  # 解析失败时绝不调 host.py
+        self.assertFalse(os.path.exists(bundle_dir))
+
+    def test_missing_key_cleans_staging_bundle(self):
+        _insert_task(self.db_path, "A")
+        self.upload_dir = os.path.join(self.tmp.name, "uploads")
+        bundle_dir = self._attach_bundle("A")
+        worker = self._worker(env={}, upload_dir=self.upload_dir)
+        result = worker.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertFalse(os.path.exists(bundle_dir))
+
     def test_custom_env_key_name_from_arg_and_config(self):
         # 参数指定的名字出现在报错里（只含名字，不含值）。
         _insert_task(self.db_path, "A")