code-to10 v6 A8 worker: worker gap tests (auto-memories, tenant skill failure, prepare-env failures, pr-job failures, run-loop drain)

(cherry picked from commit 903bd560bf2b5796daf4b86dacc29b0732513607)

cloud/tests/test_v6_a8_worker_gaps.py

@@ -0,0 +1,425 @@
+"""code-to10 v6 A8：worker 缺口补测。
+
+补测基线（v5 头 10023fb 全量覆盖率，cloud/worker.py 3402
+语句缺 338 行，90%）：本文件按“假执行器/假派发”补下列
+区间，其余缺口行以全量复测为准：
+- 2092-2095：process_auto_memories 的 extracted 项审计。
+- 2936-2944：_tenant_dirs 技能播种失败只记日志不阻塞，
+  且日志本身抛错也被吞掉。
+- 3092-3111（及 checkout 失败行）：_prepare_environment
+  的 clone/非 git/get-url/fetch/checkout 失败分支。
+- 6626-6652：_run_pr_jobs 的 PrSkipped/PrStepError/通用
+  异常三路失败写回。
+- 6977-7010：run_loop 退出排空（杀 proc、杀容器、标
+  interrupted、审计），含 project_id 为 NULL 与审计查询
+  抛错两路 _pid 回落。
+
+绝不碰（任务口径不碰项，单列）：
+- claim/goal_resume_pending 的 except-pass 分支。
+- MCP handler-error 通知分支。
+只加测试，不改生产代码；假的是执行器/派发/播种等外部
+依赖，不是被测代码本身；不加 no-coverage pragma、不 skip。
+"""
+
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    import cloud.worker as worker_mod
+    from cloud.db import connect as _db_connect
+    from cloud.pr import PrSkipped, PrStepError
+    from cloud.pr_exec import enqueue_pr_job
+    from cloud.server import init_db
+    from cloud.worker import Worker
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    import cloud.worker as worker_mod
+    from cloud.db import connect as _db_connect
+    from cloud.pr import PrSkipped, PrStepError
+    from cloud.pr_exec import enqueue_pr_job
+    from cloud.server import init_db
+    from cloud.worker import Worker
+
+
+class _FakeSync:
+    """同步假执行器：不走 Local/Docker/Cloud，只给 run_loop
+    的排空路径一个带 kill_container 的执行器替身。"""
+
+    def dispatch_task(self, task):
+        return {"status": "completed"}
+
+    def run_task(self, task):
+        return {"status": "completed"}
+
+    def kill_container(self, name):
+        raise RuntimeError("kill boom")
+
+
+class _FakeProc:
+    """pid 不存在的假进程：_kill_proc_group 在 getpgid 处即
+    返回（不真杀），wait 抛错走排空的 except 分支。"""
+
+    pid = 999999
+
+    def wait(self, timeout=None):
+        raise RuntimeError("wait boom")
+
+
+class _RaisingLogger:
+    """warning 一律抛错的假 logger：测 _tenant_dirs 里
+    “日志失败也被吞掉”的内层 except。"""
+
+    def warning(self, *args, **kwargs):
+        raise RuntimeError("log boom")
+
+    def info(self, *args, **kwargs):
+        pass
+
+    def debug(self, *args, **kwargs):
+        pass
+
+    def error(self, *args, **kwargs):
+        pass
+
+
+def _insert_task(db_path, task_id, project_id="p1", status="queued"):
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute(
+            "INSERT INTO tasks(id, project_id, prompt, status,"
+            " idempotency_key, depends_on, priority, created_by,"
+            " created_at, updated_at, last_heartbeat)"
+            " VALUES(?, ?, ?, ?, NULL, '[]', 0, 'tester', 0, 0, NULL)",
+            (task_id, project_id, "do work", status),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _row(db_path, sql):
+    conn = sqlite3.connect(db_path)
+    try:
+        return conn.execute(sql).fetchone()
+    finally:
+        conn.close()
+
+
+class WorkerGapsBase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        self.hb = os.path.join(self.tmp.name, "hb.json")
+
+    def _worker(self, **kwargs):
+        kwargs.setdefault("env", {"XCM_API_KEY": "test-placeholder"})
+        kwargs.setdefault("heartbeat_file", self.hb)
+        kwargs.setdefault("max_retries", 0)
+        kwargs.setdefault("retry_backoff_seconds", 0)
+        return Worker(self.db_path, "/fake/host.py", **kwargs)
+
+
+class ProcessAutoMemoriesTestCase(WorkerGapsBase):
+    def test_extracted_item_audited(self):
+        worker = self._worker()
+        orig = worker_mod._auto_process_due
+        worker_mod._auto_process_due = lambda db_path, **kw: [
+            {"status": "extracted", "task_id": "T1"},
+            {"status": "other", "task_id": "T2"},
+        ]
+        try:
+            results = worker.process_auto_memories()
+        finally:
+            worker_mod._auto_process_due = orig
+        self.assertEqual(len(results), 2)
+        row = _row(self.db_path,
+                   "SELECT action, detail FROM audit_log"
+                   " WHERE action = 'worker_memory_extracted'")
+        self.assertEqual(row, ("worker_memory_extracted", "T1"))
+
+    def test_sweep_failure_returns_empty(self):
+        worker = self._worker()
+        orig = worker_mod._auto_process_due
+        worker_mod._auto_process_due = lambda db_path, **kw: (_ for _ in ()).throw(
+            RuntimeError("sweep boom"))
+        try:
+            self.assertEqual(worker.process_auto_memories(), [])
+        finally:
+            worker_mod._auto_process_due = orig
+
+
+class TenantDirsSkillFailureTestCase(WorkerGapsBase):
+    def _seeded_worker(self):
+        tenants = os.path.join(self.tmp.name, "tenants")
+        template_home = os.path.join(self.tmp.name, "template-home")
+        os.makedirs(template_home, exist_ok=True)
+        with open(os.path.join(template_home, "config.toml"), "w") as f:
+            f.write("# template\n")
+        worker = self._worker(tenants_root=tenants,
+                              tenant_template_home=template_home,
+                              seed_skills=True)
+        # 预置 codex-home 配置与非空 test-repo：播种分支只测
+        # 技能失败本身，不走到 test-repo 模板复制/报错。
+        base = os.path.join(tenants, "p1")
+        os.makedirs(os.path.join(base, "codex-home"), exist_ok=True)
+        with open(os.path.join(base, "codex-home", "config.toml"),
+                  "w") as f:
+            f.write("# existing\n")
+        os.makedirs(os.path.join(base, "test-repo"), exist_ok=True)
+        with open(os.path.join(base, "test-repo", "sample.py"),
+                  "w") as f:
+            f.write("# sample\n")
+        return worker
+
+    def test_skill_failure_only_logged_not_blocking(self):
+        worker = self._seeded_worker()
+        orig = worker_mod._ensure_skills
+        worker_mod._ensure_skills = lambda home: (_ for _ in ()).throw(
+            RuntimeError("skill boom"))
+        try:
+            dirs = worker._tenant_dirs("p1")
+        finally:
+            worker_mod._ensure_skills = orig
+        self.assertTrue(os.path.isdir(dirs["test_repo"]))
+        self.assertTrue(os.path.exists(
+            os.path.join(dirs["test_repo"], "sample.py")))
+
+    def test_skill_failure_and_logger_failure_swallowed(self):
+        worker = self._seeded_worker()
+        worker.logger = _RaisingLogger()
+        orig = worker_mod._ensure_skills
+        worker_mod._ensure_skills = lambda home: (_ for _ in ()).throw(
+            RuntimeError("skill boom"))
+        try:
+            dirs = worker._tenant_dirs("p1")
+        finally:
+            worker_mod._ensure_skills = orig
+        self.assertTrue(os.path.isdir(dirs["work"]))
+
+
+class PrepareEnvironmentTestCase(WorkerGapsBase):
+    def _env_worker(self, runner):
+        repos = os.path.join(self.tmp.name, "repos")
+        return self._worker(repos_dir=repos, repos_runner=runner), repos
+
+    def test_clone_failure(self):
+        def runner(cmd):
+            return SimpleNamespace(returncode=1, stdout="",
+                                   stderr="clone boom")
+        worker, _repos = self._env_worker(runner)
+        err = worker._prepare_environment(
+            {"id": "env1", "repo_url": "https://example.invalid/x.git"})
+        self.assertIn("git clone 失败", err)
+        self.assertIn("clone boom", err)
+
+    def test_existing_dir_not_a_git_repo(self):
+        worker, repos = self._env_worker(lambda cmd: SimpleNamespace(
+            returncode=0, stdout="", stderr=""))
+        target = os.path.join(repos, "envs", "env2")
+        os.makedirs(target, exist_ok=True)
+        err = worker._prepare_environment(
+            {"id": "env2", "repo_url": "https://example.invalid/x.git"})
+        self.assertIn("不是 git 仓库", err)
+
+    def _git_repo_worker(self, behaviour):
+        worker, repos = self._env_worker(behaviour)
+        target = os.path.join(repos, "envs", "env3")
+        os.makedirs(os.path.join(target, ".git"), exist_ok=True)
+        return worker
+
+    def test_get_url_failure(self):
+        def runner(cmd):
+            return SimpleNamespace(returncode=1, stdout="",
+                                   stderr="no remote")
+        worker = self._git_repo_worker(runner)
+        err = worker._prepare_environment(
+            {"id": "env3", "repo_url": "https://example.invalid/x.git"})
+        self.assertIn("取不到 remote origin", err)
+
+    def test_origin_mismatch(self):
+        def runner(cmd):
+            if "get-url" in cmd:
+                return SimpleNamespace(
+                    returncode=0, stdout="https://other.invalid/y.git\n",
+                    stderr="")
+            return SimpleNamespace(returncode=0, stdout="", stderr="")
+        worker = self._git_repo_worker(runner)
+        err = worker._prepare_environment(
+            {"id": "env3", "repo_url": "https://example.invalid/x.git"})
+        self.assertIn("origin 与环境 repo_url 不一致", err)
+
+    def test_fetch_failure(self):
+        def runner(cmd):
+            if "get-url" in cmd:
+                return SimpleNamespace(
+                    returncode=0,
+                    stdout="https://example.invalid/x.git\n", stderr="")
+            if "fetch" in cmd:
+                return SimpleNamespace(returncode=1, stdout="",
+                                       stderr="fetch boom")
+            return SimpleNamespace(returncode=0, stdout="", stderr="")
+        worker = self._git_repo_worker(runner)
+        err = worker._prepare_environment(
+            {"id": "env3", "repo_url": "https://example.invalid/x.git"})
+        self.assertIn("git fetch origin 失败", err)
+        self.assertIn("fetch boom", err)
+
+    def test_checkout_failure_and_success(self):
+        def runner(cmd):
+            if "get-url" in cmd:
+                return SimpleNamespace(
+                    returncode=0,
+                    stdout="https://example.invalid/x.git\n", stderr="")
+            if "fetch" in cmd:
+                return SimpleNamespace(returncode=0, stdout="", stderr="")
+            if "checkout" in cmd:
+                return SimpleNamespace(returncode=1, stdout="",
+                                       stderr="no such branch")
+            return SimpleNamespace(returncode=0, stdout="", stderr="")
+        worker = self._git_repo_worker(runner)
+        env = {"id": "env3", "repo_url": "https://example.invalid/x.git",
+               "branch": "nope"}
+        err = worker._prepare_environment(env)
+        self.assertIn("git checkout nope 失败", err)
+        self.assertIn("no such branch", err)
+        # 同一 runner 改走成功 checkout：返回 None（准备完成）。
+        def ok_runner(cmd):
+            return SimpleNamespace(returncode=0,
+                                   stdout="https://example.invalid/x.git\n",
+                                   stderr="")
+        worker.repos_runner = ok_runner
+        self.assertIsNone(worker._prepare_environment(env))
+
+
+class RunPrJobsFailureTestCase(WorkerGapsBase):
+    def _pr_worker(self, failure):
+        worker = self._worker(env_key="TEST_KEY",
+                              host_name="elsewhere-host-1")
+        conn = _db_connect(self.db_path)
+        try:
+            worker._ensure_result_columns(conn)
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, prompt, status,"
+                " depends_on, priority, created_by, created_at,"
+                " updated_at, last_heartbeat) VALUES('t1', 'p1', 'x',"
+                " 'merged', '[]', 0, 'system', 0, 0, NULL)")
+            conn.commit()
+            enqueue_pr_job(conn, "t1", "p1", "elsewhere-host-1")
+        finally:
+            conn.close()
+        return worker
+
+    def _run_with_failure(self, failure):
+        worker = self._pr_worker(failure)
+        orig = worker_mod.execute_task_pr
+
+        def _fake(db_path, task_id, user_id, **kwargs):
+            raise failure
+        worker_mod.execute_task_pr = _fake
+        try:
+            results = worker._run_pr_jobs()
+        finally:
+            worker_mod.execute_task_pr = orig
+        return results
+
+    def test_pr_skipped_marks_failed(self):
+        results = self._run_with_failure(PrSkipped("no longer merged"))
+        self.assertEqual(len(results), 1)
+        self.assertEqual(results[0]["status"], "failed")
+        self.assertIn("skipped: no longer merged", results[0]["error"])
+        row = _row(self.db_path,
+                   "SELECT status FROM pr_jobs WHERE task_id = 't1'")
+        self.assertEqual(row[0], "failed")
+
+    def test_pr_step_error_marks_failed(self):
+        results = self._run_with_failure(
+            PrStepError("push", "remote denied"))
+        self.assertEqual(results[0]["error"], "push: remote denied")
+        row = _row(self.db_path,
+                   "SELECT status FROM pr_jobs WHERE task_id = 't1'")
+        self.assertEqual(row[0], "failed")
+
+    def test_generic_error_marks_failed(self):
+        results = self._run_with_failure(RuntimeError("boom"))
+        self.assertEqual(results[0]["error"], "boom")
+        row = _row(self.db_path,
+                   "SELECT status FROM pr_jobs WHERE task_id = 't1'")
+        self.assertEqual(row[0], "failed")
+
+
+class RunLoopDrainTestCase(WorkerGapsBase):
+    def _drain_worker(self, task_id, project_id):
+        _insert_task(self.db_path, task_id, project_id=project_id,
+                     status="running")
+        worker = self._worker(executor=_FakeSync())
+        worker._running_set(task_id, proc=_FakeProc(), container="c1")
+        worker.executor.kill_container = lambda name: (_ for _ in ()).throw(
+            RuntimeError("kill boom"))
+        return worker
+
+    def _run_drain(self, worker):
+        stop = threading.Event()
+        stop.set()
+        results = worker.run_loop(interval=10, drain_timeout=0,
+                                  stop_event=stop)
+        self.assertEqual(results, [])
+        return worker
+
+    def test_drain_kills_proc_and_container_marks_interrupted(self):
+        worker = self._drain_worker("t-drain", "p1")
+        self._run_drain(worker)
+        row = _row(self.db_path,
+                   "SELECT status, interrupted_reason FROM tasks"
+                   " WHERE id = 't-drain'")
+        self.assertEqual(row[0], "interrupted")
+        self.assertEqual(row[1], "worker_shutdown")
+        audit = _row(self.db_path,
+                     "SELECT action, detail FROM audit_log"
+                     " WHERE action = 'worker_shutdown_interrupted'")
+        self.assertEqual(audit,
+                         ("worker_shutdown_interrupted", "t-drain"))
+
+    def test_drain_null_project_id_audits_empty_project(self):
+        worker = self._drain_worker("t-null", None)
+        self._run_drain(worker)
+        audit = _row(self.db_path,
+                     "SELECT action, detail FROM audit_log"
+                     " WHERE action = 'worker_shutdown_interrupted'"
+                     " AND detail = 't-null'")
+        self.assertIsNotNone(audit)
+
+    def test_drain_audit_lookup_failure_still_interrupts(self):
+        worker = self._drain_worker("t-dberr", "p1")
+        orig_connect = worker_mod._db_connect
+        import inspect
+
+        def _guarded(db_path):
+            # 只有 run_loop 本体直接发起的审计查询才抛错：
+            # _set_status/_mark_interrupted_reason 调 _db_connect
+            # 时直接调用方是各自方法名，不受影响；排空审计的
+            # except 分支据此覆盖。
+            caller = inspect.currentframe().f_back
+            if caller is not None and \
+                    caller.f_code.co_name == "run_loop":
+                raise RuntimeError("db gone")
+            return orig_connect(db_path)
+        worker_mod._db_connect = _guarded
+        try:
+            self._run_drain(worker)
+        finally:
+            worker_mod._db_connect = orig_connect
+        row = _row(self.db_path,
+                   "SELECT status FROM tasks WHERE id = 't-dberr'")
+        self.assertEqual(row[0], "interrupted")
+
+
+if __name__ == "__main__":
+    unittest.main()