Merge pull request #77 from openqj/feat-staging-cleanup

fix: 提前失败路也清附件暂存

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")

cloud/worker.py

@@ -960,6 +960,22 @@ class Worker:
         bundle_dir = os.path.join(self.upload_dir, bundle_id)
         return bundle_dir if os.path.isdir(bundle_dir) else None
 
+    def _cleanup_task_seed(self, task):
+        """删任务附件暂存目录：终局清理的唯一入口。
+
+        正常终局、缺 key 门、仓库解析失败三路都要走这里——
+        后两路在 run_one 里提前返回，够不到正常终局后的清理
+        （2026-10-27 实弹：仓库解析失败的任务暂存目录残留）。
+        取消走 server 侧删，这里删不到也不报错；删失败不影响
+        任务结果本身。
+        """
+        try:
+            _seed = self._task_seed_dir(task)
+            if _seed:
+                shutil.rmtree(_seed, ignore_errors=True)
+        except Exception:
+            pass
+
     def _execute_via_process(self, task, kind="dispatch",
                              deadline_seconds=None, seed_host_dir=None):
         """进程模式执行一轮：启动、轮询、取消时杀进程组。
@@ -1880,6 +1896,9 @@ class Worker:
             self._write_execution_result(
                 task_id, error_text=error, actual_model=self.model,
                 started_at=started_at, finished_at=time.time())
+            # 缺 key 门提前返回：同样要清暂存，否则与仓库解析
+            # 失败一路同款残留（见 _cleanup_task_seed）。
+            self._cleanup_task_seed(task)
             self._audit(task["project_id"], "worker_missing_key", error)
             log_event(self.logger, "worker_missing_key", task_id=task_id,
                       project_id=task["project_id"], status="failed")
@@ -1901,6 +1920,9 @@ class Worker:
             self._write_execution_result(
                 task_id, error_text=repo_error, actual_model=self.model,
                 started_at=started_at, finished_at=time.time())
+            # 仓库解析失败提前返回：够不到正常终局后的清理，
+            # 这里必须先清暂存，否则 bundle 目录永久残留。
+            self._cleanup_task_seed(task)
             self._audit(task["project_id"], "worker_repo_failed", repo_error)
             log_event(self.logger, "worker_repo_failed", task_id=task_id,
                       project_id=task["project_id"], status="failed")
@@ -2121,14 +2143,8 @@ class Worker:
             actual_model=actual_model, started_at=started_at,
             finished_at=finished_at)
         # 附件暂存清理：任务终局后删 bundle 目录（重试走已存在
-        # 续跑，文件已在 worktree 里，不再需要暂存；取消走 server
-        # 侧删，这里删不到也不报错；删失败不影响任务结果本身）。
-        try:
-            _seed = self._task_seed_dir(task)
-            if _seed:
-                shutil.rmtree(_seed, ignore_errors=True)
-        except Exception:
-            pass
+        # 续跑，文件已在 worktree 里，不再需要暂存）。
+        self._cleanup_task_seed(task)
         with self._db_lock:
             self.usage_turns += 1
             # 成本：执行器带回上游真实用量时记 actual（input/output/