coverage85-step4a: worker run_one/_run_attempt 缺口测试（+34 项）

cloud/tests/test_coverage85_worker3.py

@@ -0,0 +1,566 @@
+"""覆盖率 78%->85% 第 4 步：cloud/worker.py run_one/_run_attempt
+剩余缺口的真实行为测试。
+
+只用临时 SQLite（server.init_db 建表）与同步假执行器：
+假执行器是执行边界（真实 Local/Docker 执行器的替身），
+绝不碰真实 sub2api/生产网关、不跑模型；超时与重试成本
+用 max_retries=0、retry_backoff_seconds=0 控制。
+"""
+
+import os
+import sqlite3
+import tempfile
+import unittest
+from unittest import mock
+
+from cloud.sandbox import DockerExecutor
+from cloud.server import init_db
+from cloud.worker import Worker
+
+
+def _insert_task(db_path, task_id, status="queued", **kw):
+    conn = sqlite3.connect(db_path)
+    try:
+        cols = {"id": task_id, "project_id": kw.pop("project_id", "proj"),
+                "prompt": kw.pop("prompt", "do it"), "status": status,
+                "created_by": kw.pop("created_by", "owner1"),
+                "created_at": kw.pop("created_at", 1000.0),
+                "updated_at": kw.pop("updated_at", 1000.0)}
+        cols.update(kw)
+        names = ", ".join(cols)
+        placeholders = ", ".join("?" * len(cols))
+        conn.execute("INSERT INTO tasks(%s) VALUES(%s)"
+                     % (names, placeholders), tuple(cols.values()))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _row(db_path, sql, params=()):
+    conn = sqlite3.connect(db_path)
+    try:
+        return conn.execute(sql, params).fetchall()
+    finally:
+        conn.close()
+
+
+def _set(db_path, sql, params=()):
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute(sql, params)
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _result(status="completed", stdout="done", **kw):
+    r = {"executor": "local", "status": status, "stdout": stdout,
+         "stderr": "", "returncode": 0, "host_status": None}
+    r.update(kw)
+    return r
+
+
+class FakeSync:
+    """同步假执行器：非 Local/Docker/Cloud，走 _run_attempt 的
+    else 分支（dispatch_task(task) / goal_task(task) 单参形）。
+    """
+
+    def __init__(self, dispatch=None, run=None, goal=None):
+        self.dispatch_result = dispatch or _result()
+        self.run_result = run or _result()
+        self.goal_result = goal or _result()
+        self.calls = []
+
+    def dispatch_task(self, task):
+        self.calls.append(("dispatch", task["id"]))
+        return dict(self.dispatch_result)
+
+    def run_task(self, task):
+        self.calls.append(("run", task["id"]))
+        return dict(self.run_result)
+
+    def goal_task(self, task):
+        self.calls.append(("goal", task["id"]))
+        return dict(self.goal_result)
+
+
+class Worker3Base(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, executor=None, **kw):
+        kw.setdefault("env", {"XCM_API_KEY": "test-placeholder"})
+        kw.setdefault("heartbeat_file", self.hb)
+        kw.setdefault("worker_id", "cov85-worker3")
+        kw.setdefault("max_retries", 0)
+        kw.setdefault("retry_backoff_seconds", 0)
+        return Worker(self.db_path, "/fake/host.py",
+                      executor=executor, **kw)
+
+
+class RunOneGatesTest(Worker3Base):
+    def test_missing_key_gate(self):
+        _insert_task(self.db_path, "t-key", status="queued")
+        fake = FakeSync()
+        w = self._worker(executor=fake, env={})
+        result = w.run_one()
+        self.assertTrue(result["dispatched"])
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("Missing environment variable", result["error"])
+        self.assertEqual(_row(self.db_path,
+                              "SELECT status FROM tasks WHERE id = 't-key'"),
+                         [("failed",)])
+        self.assertEqual(fake.calls, [])
+
+    def test_already_claimed(self):
+        _insert_task(self.db_path, "t-claim", status="running")
+        w = self._worker(executor=FakeSync())
+        task = w.get_task("t-claim")
+        result = w.run_one(task)
+        self.assertFalse(result["dispatched"])
+        self.assertEqual(result["reason"], "already_claimed")
+
+    def test_quota_gate(self):
+        _insert_task(self.db_path, "t-quota", status="queued")
+        _set(self.db_path,
+             "INSERT INTO user_quotas(user_id, quota_usd_micros,"
+             " updated_at) VALUES('owner1', 0, 0)")
+        w = self._worker(executor=FakeSync())
+        result = w.run_one()
+        self.assertTrue(result["dispatched"])
+        self.assertEqual(result["status"], "interrupted")
+        self.assertIn("quota_exceeded", result["error"])
+        self.assertEqual(_row(
+            self.db_path,
+            "SELECT interrupted_reason FROM tasks WHERE id = 't-quota'"),
+            [("quota",)])
+
+    def test_goal_cap_over_and_zero_old_form(self):
+        for rounds, want in [(101, "超过上限"), (0, "已禁用")]:
+            with self.subTest(rounds=rounds):
+                tid = "t-cap-%s" % rounds
+                _insert_task(self.db_path, tid, status="queued",
+                             verify_command="make test",
+                             goal_max_rounds=rounds)
+                w = self._worker(executor=FakeSync())
+                result = w.run_one()
+                self.assertEqual(result["status"], "failed")
+                self.assertIn(want, result["error"])
+                self.assertEqual(_row(
+                    self.db_path,
+                    "SELECT status FROM tasks WHERE id = ?", (tid,)),
+                    [("failed",)])
+
+    def test_goal_cap_bad_type_passes_gate(self):
+        # goal_max_rounds 非整数：int() 抛错后 _goal_rounds_int
+        # 为 None，门不拦，照常跑完（门只拦可解析的超限值）。
+        _insert_task(self.db_path, "t-cap-bad", status="queued",
+                     verify_command="make test", goal_max_rounds="abc")
+        w = self._worker(executor=FakeSync())
+        result = w.run_one()
+        self.assertEqual(result["status"], "completed")
+
+    def test_gateway_failure_gate(self):
+        _insert_task(self.db_path, "t-gw", status="queued")
+        w = self._worker(executor=FakeSync())
+        with mock.patch.dict(os.environ, {"CLOUD_GATEWAY": "sub2api",
+                                          "SUB2API_BASE_URL": ""}):
+            result = w.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("网关准备失败", result["error"])
+
+    def test_plain_resume_message_swaps_prompt(self):
+        _insert_task(self.db_path, "t-resume", status="queued",
+                     resume_message="继续说下去")
+        fake = FakeSync()
+        w = self._worker(executor=fake)
+        result = w.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(fake.calls, [("dispatch", "t-resume")])
+        # 续跑消息已消费：库里 resume_message 清掉。
+        self.assertEqual(_row(
+            self.db_path,
+            "SELECT resume_message FROM tasks WHERE id = 't-resume'"),
+            [(None,)])
+
+    def test_verify_resume_message_is_breakpoint_kind(self):
+        # 旧验收形态 + 续跑消息：prompt 不换、_resume_kind 为
+        # breakpoint，假执行器走 goal_task 一次即完成。
+        _insert_task(self.db_path, "t-resume-v", status="queued",
+                     verify_command="make test",
+                     resume_message="继续原目标")
+        fake = FakeSync()
+        w = self._worker(executor=fake)
+        result = w.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(fake.calls, [("goal", "t-resume-v")])
+
+    def test_goal_resume_pending_fresh_restart(self):
+        _insert_task(self.db_path, "t-grp", status="queued",
+                     goal_mode=1, goal_resume_pending=1)
+        fake = FakeSync()
+        w = self._worker(executor=fake)
+        result = w.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(fake.calls, [("goal", "t-grp")])
+        self.assertEqual(_row(
+            self.db_path,
+            "SELECT goal_resume_pending FROM tasks WHERE id = 't-grp'"),
+            [(0,)])
+
+
+class RunOneAttemptFlowTest(Worker3Base):
+    def test_completed_simple(self):
+        _insert_task(self.db_path, "t-ok", status="queued")
+        w = self._worker(executor=FakeSync())
+        result = w.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(result["stdout"], "done")
+        self.assertIsNotNone(result["cost_record_id"])
+
+    def test_failed_simple_error_text(self):
+        _insert_task(self.db_path, "t-fail", status="queued")
+        fake = FakeSync(dispatch=_result(status="failed", stdout="",
+                                         stderr="boom-err"))
+        w = self._worker(executor=fake)
+        result = w.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("boom-err", result["error"])
+
+    def test_timeout_converts_to_failed(self):
+        _insert_task(self.db_path, "t-timeout", status="queued")
+        fake = FakeSync(dispatch=_result(status="timeout", stdout=""))
+        w = self._worker(executor=fake)
+        result = w.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("task timeout", result["error"])
+
+    def test_retry_same_model(self):
+        _insert_task(self.db_path, "t-retry", status="queued")
+        fake = FakeSync(dispatch=_result(status="failed", stdout="",
+                                         stderr="first-fail"))
+        w = self._worker(executor=fake, max_retries=1,
+                         retry_backoff_seconds=0)
+        result = w.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("[retries=1]", result["error"])
+        self.assertEqual([c[0] for c in fake.calls],
+                         ["dispatch", "dispatch"])
+
+    def test_fallback_model_second_failure_combines_errors(self):
+        _insert_task(self.db_path, "t-fb", status="queued")
+        calls = {"n": 0}
+
+        class _Fb(FakeSync):
+            def dispatch_task(self, task):
+                calls["n"] += 1
+                if calls["n"] == 1:
+                    return _result(status="failed", stdout="",
+                                   stderr="primary-bad 429",
+                                   host_status="failed")
+                return _result(status="failed", stdout="",
+                               stderr="fallback-bad",
+                               host_status="failed")
+        fake = _Fb()
+        w = self._worker(executor=fake, fallback_models=["fb-model"])
+        result = w.run_one()
+        self.assertEqual(result["status"], "failed")
+        # 真 fallback 换模型再跑一次：两次原因都写进错误文本。
+        self.assertEqual(calls["n"], 2)
+        self.assertIn("primary model", result["error"])
+        self.assertIn("fallback model", result["error"])
+        self.assertIn("fallback-bad", result["error"])
+
+    def test_fallback_via_token(self):
+        _insert_task(self.db_path, "t-fb2", status="queued")
+        seq = [("failed", "model overloaded 429"),
+               ("failed", "second bad")]
+
+        class _Fb2(FakeSync):
+            def __init__(self):
+                super().__init__()
+                self.n = 0
+
+            def dispatch_task(self, task):
+                self.n += 1
+                status, err = seq[min(self.n - 1, len(seq) - 1)]
+                return _result(status=status, stdout="",
+                               stderr=err, host_status="failed")
+        fake = _Fb2()
+        w = self._worker(executor=fake, fallback_models=["fb-model"])
+        result = w.run_one()
+        self.assertEqual(result["status"], "failed")
+        # 真 fallback 走过时错误文本含两次原因（或至少第二次）。
+        self.assertEqual(fake.n, 2)
+        self.assertIn("fallback model", result["error"])
+
+    def test_cancelled_via_result(self):
+        _insert_task(self.db_path, "t-cancel", status="queued")
+        fake = FakeSync(dispatch=_result(status="cancelled", stdout=""))
+        w = self._worker(executor=fake)
+        result = w.run_one()
+        self.assertEqual(result["status"], "interrupted")
+        self.assertEqual(result["error"], "cancelled")
+
+    def test_goal_already_exists_falls_back_to_resume(self):
+        # goal 首轮 stdout 含“已存在”：run_one 走一次断点续跑，
+        # 续跑完成即最终完成（不直接 failed）。
+        _insert_task(self.db_path, "t-goal-exists", status="queued",
+                     goal_mode=1)
+        calls = []
+
+        class _GoalEx(FakeSync):
+            def goal_task(self, task):
+                calls.append(task["id"])
+                if len(calls) == 1:
+                    return _result(status="failed", stdout="任务已存在")
+                return _result(status="completed", stdout="resumed-ok")
+        w = self._worker(executor=_GoalEx())
+        result = w.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(result["stdout"], "resumed-ok")
+        self.assertEqual(len(calls), 2)
+
+    def test_goal_met_false_notifies_goal_unmet(self):
+        _insert_task(self.db_path, "t-goal-unmet", status="queued",
+                     goal_mode=1)
+        fake = FakeSync(goal=_result(status="completed", stdout="",
+                                     goal_met=False))
+        w = self._worker(executor=fake)
+        result = w.run_one()
+        self.assertEqual(result["status"], "completed")
+        actions = _row(self.db_path,
+                       "SELECT action FROM audit_log ORDER BY id")
+        self.assertIn(("worker_completed",), actions)
+
+    def test_followup_goal_convert_queues_task(self):
+        _insert_task(self.db_path, "t-fu", status="queued")
+        _set(self.db_path,
+             "INSERT INTO task_followups(id, task_id, project_id, body,"
+             " status, created_at) VALUES('fu1', 't-fu', 'proj',"
+             " '/goal 收齐周报', 'pending', 1000)")
+        w = self._worker(executor=FakeSync())
+        result = w.run_one()
+        self.assertEqual(result["status"], "queued")
+        self.assertEqual(_row(
+            self.db_path,
+            "SELECT status FROM tasks WHERE id = 't-fu'"), [("queued",)])
+        self.assertEqual(_row(
+            self.db_path,
+            "SELECT status FROM task_followups WHERE id = 'fu1'"),
+            [("done",)])
+
+    def test_followup_plain_message_runs_second_round(self):
+        _insert_task(self.db_path, "t-fu2", status="queued")
+        _set(self.db_path,
+             "INSERT INTO task_followups(id, task_id, project_id, body,"
+             " status, created_at) VALUES('fu2', 't-fu2', 'proj',"
+             " '再补充一点', 'pending', 1000)")
+        fake = FakeSync()
+        w = self._worker(executor=fake)
+        result = w.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(len(fake.calls), 2)
+        self.assertEqual(_row(
+            self.db_path,
+            "SELECT status FROM task_followups WHERE id = 'fu2'"),
+            [("done",)])
+
+    def test_followup_failure_stops_drain(self):
+        _insert_task(self.db_path, "t-fu3", status="queued")
+        _set(self.db_path,
+             "INSERT INTO task_followups(id, task_id, project_id, body,"
+             " status, created_at) VALUES('fu3', 't-fu3', 'proj',"
+             " '再补充一点', 'pending', 1000)")
+
+        class _FuFail(FakeSync):
+            def dispatch_task(self, task):
+                self.calls.append(("dispatch", task["id"]))
+                if len(self.calls) == 1:
+                    return _result()
+                return _result(status="failed", stdout="",
+                               stderr="fu-bad")
+        w = self._worker(executor=_FuFail())
+        result = w.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("fu-bad", result["error"])
+        self.assertEqual(_row(
+            self.db_path,
+            "SELECT status FROM task_followups WHERE id = 'fu3'"),
+            [("failed",)])
+
+    def test_quota_stop_after_attempt(self):
+        # 首轮返回 quota_exceeded：run_one 走配额中断收尾，
+        # 状态 interrupted、续跑消息保留（本例无消息，只断状态）。
+        _insert_task(self.db_path, "t-qstop", status="queued")
+        fake = FakeSync(dispatch=_result(status="quota_exceeded",
+                                         stdout=""))
+        w = self._worker(executor=fake)
+        result = w.run_one()
+        self.assertEqual(result["status"], "interrupted")
+        self.assertIn("quota", result["error"])
+        self.assertEqual(_row(
+            self.db_path,
+            "SELECT interrupted_reason FROM tasks WHERE id = 't-qstop'"),
+            [("quota",)])
+
+
+class RunAttemptDirectTest(Worker3Base):
+    def _task_dict(self, tid, **kw):
+        d = {"id": tid, "project_id": "proj", "prompt": "do it",
+             "created_by": "owner1"}
+        d.update(kw)
+        return d
+
+    def test_dispatch_already_exists_runs_again(self):
+        w = self._worker(executor=FakeSync(
+            dispatch=_result(status="failed", stdout="任务已存在"),
+            run=_result(stdout="run-ok")))
+        result, error = w._run_attempt(self._task_dict("t-ae"))
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(result["stdout"], "run-ok")
+        self.assertIsNone(error)
+
+    def test_executor_raises_becomes_failed(self):
+        class _Boom(FakeSync):
+            def dispatch_task(self, task):
+                raise RuntimeError("runner 炸了")
+        w = self._worker(executor=_Boom())
+        result, error = w._run_attempt(self._task_dict("t-boom"))
+        self.assertEqual(result["status"], "failed")
+        self.assertEqual(error, "runner 炸了")
+
+    def test_failed_stdout_tail_and_docker_timeout_marker(self):
+        w = self._worker(executor=FakeSync(
+            dispatch=_result(status="failed", stdout="docker run timeout")))
+        result, _ = w._run_attempt(self._task_dict("t-dto"))
+        self.assertEqual(result["status"], "timeout")
+
+    def test_quota_hit_rewrites_to_quota_exceeded(self):
+        w = self._worker(executor=FakeSync())
+        w._mark_quota_hit("t-qh")
+        result, error = w._run_attempt(self._task_dict("t-qh"))
+        self.assertEqual(result["status"], "quota_exceeded")
+        self.assertIn("配额看门狗", error)
+
+    def test_goal_breakpoint_unsupported_for_fake_executor(self):
+        w = self._worker(executor=FakeSync())
+        task = self._task_dict("t-bp", goal_mode=1)
+        task["_goal_resume_breakpoint"] = True
+        result, error = w._run_attempt(task)
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("不支持 goal 断点续跑", error)
+
+    def test_goal_task_via_fake_executor(self):
+        w = self._worker(executor=FakeSync(
+            goal=_result(stdout="goal-done")))
+        result, _ = w._run_attempt(self._task_dict("t-goal",
+                                                   goal_mode=1))
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(result["stdout"], "goal-done")
+
+    def test_no_time_limit_task(self):
+        w = self._worker(executor=FakeSync())
+        result, _ = w._run_attempt(self._task_dict(
+            "t-ntl", no_time_limit=1))
+        self.assertEqual(result["status"], "completed")
+
+
+class DockerAttemptTest(Worker3Base):
+    def _docker_worker(self, **kw):
+        class _Docker(DockerExecutor):
+            def __init__(self):
+                super().__init__(runner=lambda cmd, timeout=None: None,
+                                 env={"XCM_API_KEY": "test"})
+                self.timeout = kw.pop("timeout", None)
+
+            def container_name(self, task):
+                return "codex-t-" + task["id"]
+
+            def goal_task(self, task, **_kw):
+                return _result(stdout="docker-goal-ok")
+
+            def goal_resume_task(self, task, text, **_kw):
+                return _result(stdout="docker-resume-ok")
+
+            def goal_resume_breakpoint_task(self, task, **_kw):
+                return _result(stdout="docker-bp-ok")
+
+            def dispatch_task(self, task, **_kw):
+                return _result(stdout="docker-dispatch-ok")
+
+            def run_task(self, task, **_kw):
+                return _result(stdout="docker-run-ok")
+
+            def ensure_network(self, name):
+                return True
+
+            def network_gateway(self, network):
+                return "10.0.0.1"
+        w = self._worker(executor=_Docker(), **kw)
+        return w
+
+    def _task_dict(self, tid, **kw):
+        d = {"id": tid, "project_id": "proj", "prompt": "do it",
+             "created_by": "owner1"}
+        d.update(kw)
+        return d
+
+    def test_docker_goal_and_resume_variants(self):
+        w = self._docker_worker()
+        result, _ = w._run_attempt(self._task_dict("t-dg",
+                                                   goal_mode=1))
+        self.assertEqual(result["stdout"], "docker-goal-ok")
+        task = self._task_dict("t-dr", goal_mode=1)
+        task["_resume"] = True
+        task["_resume_kind"] = "new-goal"
+        result, _ = w._run_attempt(task)
+        self.assertEqual(result["stdout"], "docker-resume-ok")
+        task = self._task_dict("t-dbp", goal_mode=1)
+        task["_goal_resume_breakpoint"] = True
+        result, _ = w._run_attempt(task)
+        self.assertEqual(result["stdout"], "docker-bp-ok")
+        result, _ = w._run_attempt(self._task_dict("t-dd"))
+        self.assertEqual(result["stdout"], "docker-dispatch-ok")
+
+    def test_docker_context_egress_gateway_failure(self):
+        class _NoGw(DockerExecutor):
+            def __init__(self):
+                super().__init__(runner=lambda cmd, timeout=None: None,
+                                 env={"XCM_API_KEY": "test"})
+
+            def network_gateway(self, network):
+                return None
+        w = self._worker(executor=_NoGw(), egress_proxy_port=8888)
+        result, error = w._run_attempt(self._task_dict("t-dgw"))
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("egress 代理网关解析失败", error)
+
+    def test_docker_timeout_bad_type_falls_back(self):
+        # executor.timeout 非数字：min() 抛错后回落 task_timeout，
+        # 任务仍正常跑完（不炸 _run_attempt）。
+        w = self._docker_worker(timeout="bogus")
+        result, _ = w._run_attempt(self._task_dict("t-dbt"))
+        self.assertEqual(result["status"], "completed")
+
+
+class RunLoopPrJobsTest(Worker3Base):
+    def test_run_loop_zero_and_run_all_empty(self):
+        w = self._worker(executor=FakeSync())
+        self.assertEqual(w.run_loop(max_seconds=0), [])
+        self.assertEqual(w.run_all(), [])
+
+    def test_run_one_with_explicit_task_dict(self):
+        _insert_task(self.db_path, "t-exp", status="queued")
+        w = self._worker(executor=FakeSync())
+        task = w.get_task("t-exp")
+        result = w.run_one(task)
+        self.assertEqual(result["status"], "completed")
+
+
+if __name__ == "__main__":
+    unittest.main()