coverage85-step2b: server/worker 剩余缺口真实行为测试（+24 项）

- server：_dispatch_project（项目 PATCH repo_url、成员邀请/改角
  色/删除与 404/403/409 族）、_create_task 校验 400 族与合法变体
  （goal/协调员/plan/环境）、trace/agents/trace-stream（终态 SSE
  snapshot+done）、GitHub webhook 去重与坏体。
- worker：_gateway_context（xcmapi/缺 origin/缺属主/假 sub2api
  key 与协调员父属主回查）、_ensure_gateway 缓存、_docker_task_
  context（代理/租户网络失败/搜索透传）、_deliver_marker_tools
 （守卫空跑、创建派生、goal 转换、轮询快照）、协调员孤儿扫尾、
  run_one/run_all/run_loop/main 空库路径。
生产代码零改动。

cloud/tests/test_coverage85_worker2.py

@@ -0,0 +1,387 @@
+"""覆盖率 78%->85% 第 2 步：cloud/worker.py 剩余缺口的真实行为测试。
+
+只用临时 SQLite（server.init_db 建表）与假执行器/假网关：
+_gateway_context 的 sub2api 分支用模块级 _ensure_sub2api_key 替身
+（假网关，绝不碰真实 sub2api/生产网关）；_docker_task_context
+用假执行器；_deliver_marker_tools 用真实父子任务行；main/run_loop
+用空库。绝不跑模型、不建生产任务。
+"""
+
+import io
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import unittest
+from contextlib import redirect_stdout
+from pathlib import Path
+from unittest import mock
+
+try:
+    from cloud.server import init_db
+    from cloud.worker import Worker, main as worker_main
+    import cloud.worker as worker_mod
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.server import init_db
+    from cloud.worker import Worker, main as worker_main
+    import cloud.worker as worker_mod
+
+
+class _Env:
+    def __init__(self, **kw):
+        self.kw = kw
+        self.old = {}
+
+    def __enter__(self):
+        for k, v in self.kw.items():
+            self.old[k] = os.environ.get(k)
+            if v is None:
+                os.environ.pop(k, None)
+            else:
+                os.environ[k] = v
+        return self
+
+    def __exit__(self, *exc):
+        for k, v in self.old.items():
+            if v is None:
+                os.environ.pop(k, None)
+            else:
+                os.environ[k] = v
+        return False
+
+
+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 _rows(db_path, sql, params=()):
+    conn = sqlite3.connect(db_path)
+    try:
+        return conn.execute(sql, params).fetchall()
+    finally:
+        conn.close()
+
+
+class Worker2Base(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")
+        self.worker = Worker(self.db_path, "/fake/host.py",
+                             env={"XCM_API_KEY": "test-placeholder"},
+                             heartbeat_file=self.hb,
+                             worker_id="cov85-worker2")
+
+
+class GatewayContextTest(Worker2Base):
+    def test_xcmapi_returns_none_none(self):
+        with _Env(CLOUD_GATEWAY="xcmapi", SUB2API_BASE_URL=None):
+            self.assertEqual(
+                self.worker._gateway_context({"created_by": "owner1"}),
+                (None, None))
+        with _Env(CLOUD_GATEWAY=None, SUB2API_BASE_URL=None):
+            self.assertEqual(
+                self.worker._gateway_context({"created_by": "owner1"}),
+                (None, None))
+
+    def test_sub2api_missing_origin_raises(self):
+        with _Env(CLOUD_GATEWAY="sub2api", SUB2API_BASE_URL=None):
+            with self.assertRaises(RuntimeError) as ctx:
+                self.worker._gateway_context({"created_by": "owner1"})
+            self.assertIn("SUB2API_BASE_URL", str(ctx.exception))
+
+    def test_sub2api_missing_owner_raises(self):
+        with _Env(CLOUD_GATEWAY="sub2api",
+                  SUB2API_BASE_URL="https://gw.example.test"):
+            with self.assertRaises(RuntimeError) as ctx:
+                self.worker._gateway_context({})
+            self.assertIn("created_by", str(ctx.exception))
+
+    def test_sub2api_fake_gateway_and_parent_owner(self):
+        _insert_task(self.db_path, "parent1", status="completed",
+                     created_by="owner9")
+        calls = []
+        with _Env(CLOUD_GATEWAY="sub2api",
+                  SUB2API_BASE_URL="https://gw.example.test:8443"):
+            with mock.patch.object(
+                    worker_mod, "_ensure_sub2api_key",
+                    side_effect=lambda conn, owner: (
+                        calls.append(owner), "fake-key-123")[1]):
+                provider, env = self.worker._gateway_context(
+                    {"created_by": "owner1"})
+                self.assertEqual(provider, "sub2api")
+                self.assertEqual(env["SUB2API_API_KEY"], "fake-key-123")
+                self.assertEqual(env["SUB2API_BASE_URL"],
+                                 "https://gw.example.test:8443")
+                self.assertIn("gw.example.test", env["NO_PROXY"])
+                self.assertEqual(env["NO_PROXY"], env["no_proxy"])
+                self.assertEqual(calls, ["owner1"])
+                # 协调员子任务无 created_by 时跟父任务属主走。
+                provider, env = self.worker._gateway_context(
+                    {"parent_task_id": "parent1"})
+                self.assertEqual(provider, "sub2api")
+                self.assertEqual(calls, ["owner1", "owner9"])
+
+    def test_sub2api_default_model_env_and_missing_key(self):
+        with _Env(CLOUD_SUB2API_DEFAULT_MODEL="model-x"):
+            self.assertEqual(
+                self.worker._sub2api_default_model({}), "model-x")
+        with _Env(CLOUD_SUB2API_DEFAULT_MODEL=None,
+                  SUB2API_BASE_URL=None):
+            with self.assertRaises(RuntimeError):
+                self.worker._sub2api_default_model({})
+
+    def test_ensure_gateway_caches(self):
+        task = {"created_by": "owner1"}
+        with _Env(CLOUD_GATEWAY="xcmapi"):
+            first = self.worker._ensure_gateway(task)
+            self.assertEqual(first, (None, None))
+            self.assertIn("_gateway_provider", task)
+            second = self.worker._ensure_gateway(task)
+            self.assertEqual(second, first)
+
+
+class FakeExecutor:
+    def __init__(self, gateway=None, network_ok=True):
+        self.gateway = gateway
+        self.network_ok = network_ok
+        self.networks = []
+
+    def ensure_network(self, name):
+        self.networks.append(name)
+        return self.network_ok
+
+    def network_gateway(self, network):
+        return self.gateway
+
+
+class DockerTaskContextTest(Worker2Base):
+    def test_plain_no_tenant_no_proxy(self):
+        ok, mounts, network, proxy, err = self.worker._docker_task_context(
+            {"project_id": "proj"})
+        self.assertEqual((ok, mounts, network, proxy, err),
+                         (True, None, None, None, None))
+
+    def test_egress_proxy_gateway(self):
+        w = Worker(self.db_path, "/fake/host.py",
+                   env={"XCM_API_KEY": "test-placeholder"},
+                   heartbeat_file=self.hb, worker_id="cov85-w2",
+                   egress_proxy_port=8888,
+                   executor=FakeExecutor(gateway="10.0.0.1"))
+        ok, _, _, proxy, err = w._docker_task_context({"project_id": "p"})
+        self.assertTrue(ok)
+        self.assertIsNone(err)
+        self.assertEqual(proxy["HTTP_PROXY"], "http://10.0.0.1:8888")
+        self.assertEqual(proxy["NO_PROXY"], "localhost,127.0.0.1")
+        w2 = Worker(self.db_path, "/fake/host.py",
+                    env={"XCM_API_KEY": "test-placeholder"},
+                    heartbeat_file=self.hb, worker_id="cov85-w2",
+                    egress_proxy_port=8888,
+                    executor=FakeExecutor(gateway=None))
+        ok, _, _, proxy, err = w2._docker_task_context({"project_id": "p"})
+        self.assertFalse(ok)
+        self.assertIsNone(proxy)
+        self.assertIn("egress 代理网关解析失败", err)
+
+    def test_tenant_network_failure_and_search_passthrough(self):
+        tenants = os.path.join(self.tmp.name, "tenants")
+        template_home = os.path.join(self.tmp.name, "template-home")
+        os.makedirs(template_home)
+        with open(os.path.join(template_home, "config.toml"), "w") as f:
+            f.write("[mcp_servers]\n")
+        template_repo = os.path.join(self.tmp.name, "template-repo")
+        os.makedirs(template_repo)
+        with open(os.path.join(template_repo, "sample.py"), "w") as f:
+            f.write("x = 1\n")
+        w = Worker(self.db_path, "/fake/host.py",
+                   env={"XCM_API_KEY": "test-placeholder"},
+                   heartbeat_file=self.hb, worker_id="cov85-w2",
+                   tenant_isolation=True, tenants_root=tenants,
+                   tenant_template_home=template_home,
+                   tenant_template_repo=template_repo,
+                   executor=FakeExecutor(network_ok=False))
+        ok, _, _, _, err = w._docker_task_context({"project_id": "proj"})
+        self.assertFalse(ok)
+        self.assertIn("租户 Docker 网络创建失败", err)
+        w2 = Worker(self.db_path, "/fake/host.py",
+                    env={"XCM_API_KEY": "test-placeholder"},
+                    heartbeat_file=self.hb, worker_id="cov85-w2",
+                    executor=FakeExecutor())
+        with _Env(SEARXNG_URL="http://search.example.test:8080"):
+            ok, _, _, proxy, err = w2._docker_task_context(
+                {"project_id": "proj"})
+        self.assertTrue(ok)
+        self.assertEqual(proxy["SEARXNG_URL"],
+                         "http://search.example.test:8080")
+
+    def test_tenant_safe_name_errors(self):
+        with self.assertRaises(ValueError):
+            self.worker._tenant_safe_name(None)
+        with self.assertRaises(ValueError):
+            self.worker._tenant_safe_name("...")
+        self.assertEqual(self.worker._tenant_network("proj"),
+                         "codex-proj")
+
+
+class MarkerToolsTest(Worker2Base):
+    def _followups(self, task_id):
+        return _rows(self.db_path,
+                     "SELECT body FROM task_followups WHERE task_id = ?",
+                     (task_id,))
+
+    def test_guards_are_noops(self):
+        _insert_task(self.db_path, "parent1", status="queued")
+        parent = self.worker.get_task("parent1")
+        for result, patch in [
+                (None, {}),
+                ({"status": "failed"}, {}),
+                ({"status": "completed", "task_creates": [
+                    {"prompt": "x", "name": "c", "model": None,
+                     "effort": None, "executor": None}]},
+                 {"verify_command": "make test"}),
+                ({"status": "completed", "task_creates": [
+                    {"prompt": "x", "name": "c", "model": None,
+                     "effort": None, "executor": None}]},
+                 {"goal_mode": 1}),
+                ({"status": "completed", "task_creates": [
+                    {"prompt": "x", "name": "c", "model": None,
+                     "effort": None, "executor": None}]},
+                 {"coordinator_parts": [{"name": "a"}]}),
+                ({"status": "completed"}, {}),
+        ]:
+            with self.subTest(patch=patch):
+                task = dict(parent)
+                task.update(patch)
+                self.worker._deliver_marker_tools(task, result)
+        self.assertEqual(self._followups("parent1"), [])
+        self.assertEqual(_rows(self.db_path,
+                               "SELECT COUNT(*) FROM tasks"), [(1,)])
+
+    def test_creates_child_and_report(self):
+        _insert_task(self.db_path, "parent1", status="completed")
+        parent = self.worker.get_task("parent1")
+        self.worker._deliver_marker_tools(parent, {
+            "status": "completed",
+            "task_creates": [{"prompt": "child do", "name": "c1",
+                              "model": None, "effort": None,
+                              "executor": None}]})
+        children = _rows(
+            self.db_path,
+            "SELECT prompt, status, parent_task_id FROM tasks "
+            "WHERE parent_task_id = ?", ("parent1",))
+        self.assertEqual(children, [("child do", "queued", "parent1")])
+        followups = self._followups("parent1")
+        self.assertEqual(len(followups), 1)
+        self.assertIn("任务工具报告", followups[0][0])
+
+    def test_goal_convert_queues_goal_followup(self):
+        _insert_task(self.db_path, "parent1", status="completed")
+        parent = self.worker.get_task("parent1")
+        self.worker._deliver_marker_tools(parent, {
+            "status": "completed",
+            "goal_converts": [{"goal": "收齐周报"}]})
+        followups = self._followups("parent1")
+        self.assertEqual(len(followups), 1)
+        self.assertIn("/goal", followups[0][0])
+        self.assertIn("收齐周报", followups[0][0])
+
+    def test_poll_snapshot_report(self):
+        _insert_task(self.db_path, "parent1", status="completed")
+        _insert_task(self.db_path, "child1", status="completed")
+        parent = self.worker.get_task("parent1")
+        self.worker._deliver_marker_tools(parent, {
+            "status": "completed",
+            "task_polls": [{"task_id": "child1", "interval_seconds": 0.05,
+                            "timeout_seconds": 5, "chars": 200}]})
+        followups = self._followups("parent1")
+        self.assertEqual(len(followups), 1)
+        self.assertIn("child1", followups[0][0])
+
+
+class OrphanSweepTest(Worker2Base):
+    def test_sweep_marks_orphans_only(self):
+        _insert_task(self.db_path, "p-done", status="completed")
+        _insert_task(self.db_path, "p-run", status="running")
+        _insert_task(self.db_path, "c1", status="queued",
+                     parent_task_id="p-done")
+        _insert_task(self.db_path, "c2", status="queued",
+                     parent_task_id="p-run")
+        _insert_task(self.db_path, "c3", status="queued",
+                     parent_task_id="p-missing")
+        _insert_task(self.db_path, "c4", status="failed",
+                     parent_task_id="p-done")
+        marked = self.worker.sweep_orphaned_coordinator_children()
+        self.assertEqual(sorted(marked),
+                         [("p-done", "c1"), ("p-missing", "c3")])
+        statuses = dict(_rows(
+            self.db_path, "SELECT id, status FROM tasks"))
+        self.assertEqual(statuses["c1"], "interrupted")
+        self.assertEqual(statuses["c3"], "interrupted")
+        self.assertEqual(statuses["c2"], "queued")
+        self.assertEqual(statuses["c4"], "failed")
+        reasons = _rows(
+            self.db_path,
+            "SELECT interrupted_reason FROM tasks WHERE id = 'c1'")
+        self.assertEqual(reasons, [("coordinator_orphan",)])
+
+
+class RunLoopMainTest(Worker2Base):
+    def test_run_one_no_ready_and_budget(self):
+        result = self.worker.run_one()
+        self.assertEqual(result, {"dispatched": False,
+                                  "reason": "no_ready_task"})
+        self.assertEqual(self.worker.run_all(), [])
+        tight = Worker(self.db_path, "/fake/host.py",
+                       env={"XCM_API_KEY": "test-placeholder"},
+                       heartbeat_file=self.hb, worker_id="cov85-w2",
+                       max_cost_usd=0)
+        result = tight.run_one()
+        self.assertEqual(result["dispatched"], False)
+        self.assertEqual(result["reason"], "budget_exceeded")
+
+    def test_run_loop_max_seconds_zero(self):
+        results = self.worker.run_loop(max_seconds=0)
+        self.assertEqual(results, [])
+        self.assertTrue(os.path.exists(self.hb))
+
+    def test_main_single_and_loop_and_bad_executor(self):
+        out = io.StringIO()
+        with redirect_stdout(out):
+            worker_main(["--db", self.db_path,
+                         "--host-py", "/fake/host.py",
+                         "--heartbeat-file", self.hb,
+                         "--worker-id", "cov85-main"])
+        self.assertEqual(json.loads(out.getvalue()), [])
+        out = io.StringIO()
+        with redirect_stdout(out):
+            worker_main(["--db", self.db_path,
+                         "--host-py", "/fake/host.py",
+                         "--heartbeat-file", self.hb,
+                         "--worker-id", "cov85-main",
+                         "--loop", "--max-seconds", "0"])
+        self.assertEqual(json.loads(out.getvalue()), [])
+        with self.assertRaises(SystemExit) as ctx:
+            worker_main(["--db", self.db_path, "--executor", "bogus"])
+        self.assertEqual(ctx.exception.code, 2)
+
+
+if __name__ == "__main__":
+    unittest.main()