残留批次A：成员删除、PR needs_local_worker 落库、推送 DNS TOCTOU

- 成员删除：DELETE /projects/{id}/members/{user_id}，限 owner 或
  全局管理员，禁删唯一 owner，记 remove_member 审计
  （AuthStore.remove_member + server 分发，沿用 PATCH 守卫口径）
- PR needs_local_worker 落库：非 local 扫到 queued 任务时 CAS
  写行状态为 needs_local_worker；claim/find_open 同步认该状态，
  标记过的行仍可被 local worker 认领、不被搁置；另修
  claim/find_open 对裸 sqlite3 tuple 行的 dict() 兜底
  （TypeError/ValueError）
- 推送 DNS TOCTOU：发送改走建连时解析+判段的连接类，解析
  一次、只拨判过的 IP，Host/SNI 仍为原域名；opener 直接构造
  以排除 ProxyHandler 改写（build_opener 默认会塞代理）

cloud/auth.py

@@ -307,6 +307,25 @@ class AuthStore:
             "%s:%s" % (target_user_id, role),
         )
 
+    def remove_member(self, project_id, actor_user_id, target_user_id):
+        """删成员：只有 owner 能删，否则 raise PermissionError；同时记审计。"""
+        if self._member_role(project_id, actor_user_id) != "owner":
+            raise PermissionError(
+                "only project owner can change members"
+            )
+        self.conn.execute(
+            "DELETE FROM project_members"
+            " WHERE project_id = ? AND user_id = ?",
+            (project_id, target_user_id),
+        )
+        self.conn.commit()
+        self.audit(
+            actor_user_id,
+            project_id,
+            "remove_member",
+            target_user_id,
+        )
+
     def check_permission(self, user_id, project_id, action):
         """检查权限：满足最低角色返回 True，否则 False（含未加入项目）。"""
         min_role = _ACTION_MIN_ROLE.get(action)

cloud/pr_exec.py

@@ -54,18 +54,23 @@ def ensure_pr_jobs_table(conn):
 
 
 def find_open_pr_job(conn, task_id):
-    """找该任务未完成的 pr 任务（去重：同一任务同时只排一个）。"""
+    """找该任务未完成的 pr 任务（去重：同一任务同时只排一个）。
+
+    needs_local_worker 也算未完成：非 local worker 标记过的任务
+    仍在等 local worker 认领，再调 /pr 必须复用它，不能重排。"""
     ensure_pr_jobs_table(conn)
     row = conn.execute(
         "SELECT * FROM pr_jobs WHERE task_id = ?"
-        " AND status IN ('queued', 'running')"
+        " AND status IN ('queued', 'running', 'needs_local_worker')"
         " ORDER BY created_at ASC LIMIT 1",
         (task_id,)).fetchone()
     if row is None:
         return None
     try:
         return dict(row)
-    except TypeError:
+    except (TypeError, ValueError):
+        # 裸 sqlite3 连接（无 Row 工厂）返回 tuple：dict() 对
+        # tuple 抛 ValueError（对 Row 抛 TypeError 的旧路径不动）。
         return {"id": row[0], "task_id": row[1], "project_id": row[2],
                 "host": row[3], "status": row[4]}
 
@@ -84,34 +89,63 @@ def enqueue_pr_job(conn, task_id, project_id, host):
 
 
 def claim_pr_job(conn, host):
-    """CAS 认领本主机的一个 queued pr 任务（最老优先）。
+    """CAS 认领本主机的一个 pr 任务（最老优先）。
 
-    两个同主机 worker 抢同一任务时只有一个能拿到 rowcount == 1。
+    认领 queued 与 needs_local_worker 两种状态：后者是非 local
+    worker 标记的“需 local worker”任务，仍在等 local worker，
+    必须能被认领（否则标记等于搁置）。两个同主机 worker 抢
+    同一任务时只有一个能拿到 rowcount == 1。
     返回任务 dict 或 None。
     """
     ensure_pr_jobs_table(conn)
     row = conn.execute(
-        "SELECT * FROM pr_jobs WHERE host = ? AND status = 'queued'"
+        "SELECT * FROM pr_jobs WHERE host = ?"
+        " AND status IN ('queued', 'needs_local_worker')"
         " ORDER BY created_at ASC LIMIT 1",
         (host,)).fetchone()
     if row is None:
         return None
-    job_id = row["id"] if hasattr(row, "__getitem__") else row[0]
+    try:
+        job_id = row["id"]
+    except TypeError:
+        # 裸 sqlite3 连接（无 Row 工厂）返回 tuple：按位置取。
+        # hasattr(__getitem__) 判据对 tuple 也为真，不能用它区分。
+        job_id = row[0]
     now = time.time()
     cur = conn.execute(
         "UPDATE pr_jobs SET status = 'running', started_at = ?,"
-        " updated_at = ? WHERE id = ? AND status = 'queued'",
+        " updated_at = ? WHERE id = ?"
+        " AND status IN ('queued', 'needs_local_worker')",
         (now, now, job_id))
     conn.commit()
     if cur.rowcount != 1:
         return None
     try:
         return dict(row)
-    except TypeError:
+    except (TypeError, ValueError):
+        # 同上：tuple 行的 dict() 抛 ValueError。
         return {"id": row[0], "task_id": row[1], "project_id": row[2],
                 "host": row[3], "status": "running"}
 
 
+def mark_pr_job_needs_local_worker(conn, job_id, reason):
+    """把 queued 的 pr 任务标为 needs_local_worker（CAS）。
+
+    只有仍是 queued 时才改写并返回 True；已被 local worker
+    认领（running）或已终态时返回 False，绝不覆盖。标记后
+    任务仍可被 local worker 认领（见 claim_pr_job），只是
+    重读时能看到它在等 local worker，而不是“没人处理”。
+    """
+    ensure_pr_jobs_table(conn)
+    now = time.time()
+    cur = conn.execute(
+        "UPDATE pr_jobs SET status = 'needs_local_worker', error = ?,"
+        " updated_at = ? WHERE id = ? AND status = 'queued'",
+        (reason, now, job_id))
+    conn.commit()
+    return cur.rowcount == 1
+
+
 def finish_pr_job(conn, job_id, status, pr_url=None, pr_number=None,
                   error=None):
     """写回 pr 任务终态（completed/failed）。"""

cloud/push.py

@@ -46,6 +46,8 @@
   密钥、token、证书一律不由本模块读取、写入或返回。
 """
 
+import errno
+import http.client
 import ipaddress
 import json
 import os
@@ -285,13 +287,16 @@ def _resolve_channel_sender(sender, channel):
 _PUSH_SEND_TIMEOUT = 10
 
 
-def _post_json(url, payload):
+def _post_json(url, payload, resolver=None):
     """POST JSON 到 url：返回 (ok, detail)，2xx 为 ok。
 
-    HTTP 状态错误与网络异常一律返回 (False, reason)，不抛
-    （调用方据此记 errors）；只有调用方传了非法 URL 时
-    Request 构造本身可能抛 ValueError，由发送端先校验 scheme
-    挡住，不走到这里。
+    建连时解析一次并判段、钉 IP 直连（见 _checked_opener）：
+    Host 头与 TLS SNI 仍为原域名，只把拨号地址换成判过的
+    IP——预检解析与建连解析不一致时，建连侧的判定为准，
+    绝不会拨向未验证的 IP。HTTP 状态错误与网络异常一律
+    返回 (False, reason)，不抛（调用方据此记 errors）；
+    只有调用方传了非法 URL 时 Request 构造本身可能抛
+    ValueError，由发送端先校验 scheme 挡住，不走到这里。
     """
     data = json.dumps(payload).encode("utf-8")
     request = urllib.request.Request(
@@ -299,7 +304,7 @@ def _post_json(url, payload):
         headers={"Content-Type": "application/json",
                  "User-Agent": "codex-app-server"})
     try:
-        with urllib.request.urlopen(
+        with _checked_opener(resolver).open(
                 request, timeout=_PUSH_SEND_TIMEOUT) as response:
             response.read()
             return True, "HTTP %s" % response.status
@@ -309,6 +314,114 @@ def _post_json(url, payload):
         return False, "%s: %s" % (type(exc).__name__, exc)
 
 
+def _pick_connect_ip(host, resolver):
+    """建连时解析一次并判段，返回要拨的 IP（只抛 OSError）。
+
+    字面量 IP 不经过 DNS（只复核段）；域名解析一次：
+    allow-private 时取首个，否则取首个全局单播 IP；无可用
+    IP 时抛 OSError。调用方拨的就是这里判过的 IP，之后
+    不再二次解析——校验与建连之间的 DNS 重绑定窗口就此
+    闭合（旧版 validate 后 urlopen 再解析一次的 TOCTOU）。
+    """
+    if _allow_private_targets():
+        try:
+            ipaddress.ip_address(host)
+            return host
+        except ValueError:
+            pass
+        ips = (resolver or _default_resolver)(host)
+        if not ips:
+            raise OSError("DNS resolution failed for target host")
+        return ips[0]
+    try:
+        ipaddress.ip_address(host)
+    except ValueError:
+        pass
+    else:
+        if _ip_is_blocked(host):
+            raise OSError(
+                "target resolves to a non-public address: %s" % host)
+        return host
+    ips = (resolver or _default_resolver)(host)
+    if not ips:
+        raise OSError("DNS resolution failed for target host")
+    for ip_text in ips:
+        if not _ip_is_blocked(ip_text):
+            return ip_text
+    raise OSError("target host resolves to a non-public address")
+
+
+def _checked_connection_class(base, resolver):
+    """建连时解析+判段的连接类：Host 头与 TLS SNI 仍为原域名
+    （self.host 不动），只把拨号地址换成判过的 IP。
+
+    resolver 由调用方注入（测试 mock DNS 用）；为 None 时
+    用默认解析。代理隧道场景（_tunnel_host 已置）仍走基类：
+    隧道里解析权在代理方，钉 IP 不适用，预检判定为准。
+    """
+    class CheckedConnection(base):
+        def connect(self):
+            if getattr(self, "_tunnel_host", None):
+                super().connect()
+                return
+            picked = _pick_connect_ip(self.host, resolver)
+            self.sock = socket.create_connection(
+                (picked, self.port), self.timeout, self.source_address)
+            try:
+                self.sock.setsockopt(
+                    socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+            except OSError as exc:
+                if exc.errno != errno.ENOPROTOOPT:
+                    raise
+            if isinstance(self, http.client.HTTPSConnection):
+                self.sock = self._context.wrap_socket(
+                    self.sock, server_hostname=self.host)
+    return CheckedConnection
+
+
+class _CheckedHTTPHandler(urllib.request.HTTPHandler):
+    def __init__(self, resolver=None):
+        super().__init__()
+        self._resolver = resolver
+
+    def http_open(self, req):
+        return self.do_open(
+            _checked_connection_class(http.client.HTTPConnection,
+                                      self._resolver), req)
+
+
+class _CheckedHTTPSHandler(urllib.request.HTTPSHandler):
+    def __init__(self, resolver=None, debuglevel=0, context=None,
+                 check_hostname=None):
+        super().__init__(debuglevel=debuglevel, context=context,
+                         check_hostname=check_hostname)
+        self._resolver = resolver
+
+    def https_open(self, req):
+        return self.do_open(
+            _checked_connection_class(http.client.HTTPSConnection,
+                                      self._resolver), req,
+            context=self._context, check_hostname=self._check_hostname)
+
+
+def _checked_opener(resolver=None):
+    """发送用 opener：建连时解析+判段的连接类 + 重定向/错误处理。
+
+    刻意直接构造 OpenerDirector、不用 build_opener：后者会
+    自动塞入 ProxyHandler，本机 egress 代理会把请求改写成
+    代理端口、钉 IP 失效（实测拨号跑到代理端口）。钉 IP 的
+    前提是直连；代理环境下发送会直连失败并记入 errors
+    （loud，不静默成功）。重定向仍跟随，但每一跳都重新
+    走建连判定（新目标同样解析+判段）。
+    """
+    opener = urllib.request.OpenerDirector()
+    opener.add_handler(_CheckedHTTPHandler(resolver))
+    opener.add_handler(_CheckedHTTPSHandler(resolver))
+    opener.add_handler(urllib.request.HTTPRedirectHandler())
+    opener.add_handler(urllib.request.HTTPErrorProcessor())
+    return opener
+
+
 def _require_http_url(target, channel):
     """发送端 target 校验：只接受 http(s) URL，其他一律 ValueError
     （dispatch 记入 errors，不静默成功、不乱 POST）。"""
@@ -358,8 +471,11 @@ def validate_push_target(target, resolver=None, check_dns=True):
 
     顺序：scheme（http/https）-> 主机名存在 -> 元数据主机名 ->
     IP 字面量直接判段 -> （check_dns 时）DNS 解析后逐 IP 判段，
-    解析失败或解析出空结果都拒绝（DNS 重绑定下解析时与发送时
-    仍可能不一致，这是本层能做的上限，见残留说明）。
+    解析失败或解析出空结果都拒绝。本预检只是 fail-fast：
+    真正起强制作用的是发送时建连类的解析+判段（解析一次、
+    只拨判过的 IP），预检解析与建连解析不一致时以建连侧
+    的判定为准，不会拨向未验证的 IP（旧版 validate 后
+    urlopen 再解析一次的 TOCTOU 已闭合）。
     check_dns=False 时只做字面量判定：供直接调 send_* 但不想
     做 DNS 的调用方用；dispatch 的真实发送路径一律
     check_dns=True，显式 sender 路径不走本预检（见 dispatch）。
@@ -424,7 +540,7 @@ def send_webhook(message):
         "project_id": message["project_id"],
         "task_id": payload.get("task_id"),
         "payload": message["payload"],
-    })
+    }, resolver=message.get("_resolver"))
     if not ok:
         raise RuntimeError("webhook delivery failed: %s" % detail)
 
@@ -450,7 +566,8 @@ def send_slack(message):
         message["project_id"])
     if payload.get("error"):
         text += "\nerror: %s" % str(payload["error"])[:500]
-    ok, detail = _post_json(target, {"text": text})
+    ok, detail = _post_json(target, {"text": text},
+                            resolver=message.get("_resolver"))
     if not ok:
         raise RuntimeError("slack delivery failed: %s" % detail)
 

cloud/server.py

@@ -2471,6 +2471,54 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         self._send_json(200, {"project_id": project_id,
                               "user_id": target, "role": role})
 
+    def _project_member_delete(self, user_id, project_id, target):
+        """DELETE /projects/{id}/members/{user_id}：限项目 owner
+        或全局管理员。删唯一 owner 时 409（否则项目将无 owner
+        可管）；目标非成员时 404。成功记 remove_member 审计。"""
+        if not self._project_exists(project_id):
+            self._send_json(404, {"error": "project not found"})
+            return
+        if not self._can_manage_members(user_id, project_id):
+            self._send_json(403, {"error": "forbidden"})
+            return
+        with _auth_store(self.server.db_path) as store:
+            cur = store.conn.execute(
+                "SELECT role FROM project_members"
+                " WHERE project_id = ? AND user_id = ?",
+                (project_id, target)).fetchone()
+            if cur is None:
+                self._send_json(404, {"error": "member not found"})
+                return
+            if cur[0] == "owner":
+                owners = store.conn.execute(
+                    "SELECT COUNT(*) FROM project_members"
+                    " WHERE project_id = ? AND role = 'owner'",
+                    (project_id,)).fetchone()[0]
+                if owners <= 1:
+                    self._send_json(
+                        409,
+                        {"error": "cannot remove the only owner"})
+                    return
+            self._remove_member_row(store, project_id, user_id, target)
+        self._send_json(200, {"project_id": project_id,
+                              "user_id": target})
+
+    def _remove_member_row(self, store, project_id, actor_id, target_id):
+        """删成员行并记审计：owner 走 AuthStore.remove_member
+        （内含 owner 校验与审计）；非成员的全局管理员走直接删
+        + 同格式审计（remove_member 的 owner 校验会挡住全局
+        管理员，故分流，与 _set_member_row 同款）。"""
+        if store.check_permission(actor_id, project_id,
+                                  "manage_project"):
+            store.remove_member(project_id, actor_id, target_id)
+            return
+        store.conn.execute(
+            "DELETE FROM project_members"
+            " WHERE project_id = ? AND user_id = ?",
+            (project_id, target_id))
+        store.conn.commit()
+        store.audit(actor_id, project_id, "remove_member", target_id)
+
     # -- 项目子资源：记忆 / 知识库 / 成本 -------------------------------
     def _project_exists(self, project_id):
         conn = self._db()
@@ -2510,10 +2558,12 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             else:
                 self._send_json(404, {"error": "not found"})
             return
-        # /projects/{id}/members/{user_id}：PATCH 改角色
+        # /projects/{id}/members/{user_id}：PATCH 改角色 / DELETE 删成员
         if len(parts) == 3 and parts[1] == "members":
             if method == "PATCH":
                 self._project_member_patch(user_id, project_id, parts[2])
+            elif method == "DELETE":
+                self._project_member_delete(user_id, project_id, parts[2])
             else:
                 self._send_json(404, {"error": "not found"})
             return

cloud/tests/test_member_delete.py

@@ -0,0 +1,161 @@
+"""成员删除端点单测（残留批次 A 任务 1）。
+
+DELETE /projects/{id}/members/{user_id}：
+- 限项目 owner 或全局管理员，其余 403；
+- 目标非成员 404，项目不存在 404；
+- 删唯一 owner 时 409（owner 与全局管理员均受此限）；
+- 成功后成员列表无此人，且记 remove_member 审计。
+
+测试用的 key、用户 id 全是假值；不联网、不碰生产。
+"""
+
+import http.client
+import json
+import os
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+
+
+def _req(port, method, path, body=None, key=None):
+    conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10)
+    headers = {"Content-Type": "application/json"}
+    if key:
+        headers["X-API-Key"] = key
+    conn.request(method, path,
+                 json.dumps(body) if body is not None else None, headers)
+    resp = conn.getresponse()
+    raw = resp.read()
+    conn.close()
+    try:
+        data = json.loads(raw)
+    except ValueError:
+        data = None
+    return resp.status, data
+
+
+class MemberDeleteTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.trace_dir = os.path.join(self.tmp.name, "host-traces")
+        os.makedirs(self.trace_dir, exist_ok=True)
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.maint_id, self.maint_key = self.auth.create_user("maint")
+        self.member_id, self.member_key = self.auth.create_user("member")
+        self.admin_id, self.admin_key = self.auth.create_user("admin")
+        self.outsider_id, self.out_key = self.auth.create_user("outsider")
+        self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS")
+        os.environ["CODEX_GLOBAL_ADMINS"] = self.admin_id
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.server.trace_dir = self.trace_dir
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, proj = _req(self.port, "POST", "/projects",
+                            {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201, proj)
+        self.project_id = proj["id"]
+        self.auth.set_member(self.project_id, self.owner_id,
+                             self.maint_id, "maintainer")
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+        if self._old_admins is None:
+            os.environ.pop("CODEX_GLOBAL_ADMINS", None)
+        else:
+            os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins
+
+    def _delete(self, target, key):
+        return _req(self.port, "DELETE",
+                    "/projects/%s/members/%s" % (self.project_id, target),
+                    key=key)
+
+    def _invite(self, target, role="contributor", key=None):
+        return _req(self.port, "POST",
+                    "/projects/%s/members" % self.project_id,
+                    {"user_id": target, "role": role},
+                    key or self.owner_key)
+
+    def _member_ids(self, key=None):
+        status, data = _req(self.port, "GET",
+                            "/projects/%s/members" % self.project_id,
+                            key=key or self.owner_key)
+        self.assertEqual(status, 200, data)
+        return {m["user_id"] for m in data["members"]}
+
+    def _audit_actions(self):
+        return [(a["action"], a["detail"])
+                for a in self.auth.list_audit(self.project_id)]
+
+    def test_owner_delete_member_and_audit(self):
+        self.assertEqual(self._invite(self.member_id)[0], 201)
+        status, data = self._delete(self.member_id, self.owner_key)
+        self.assertEqual(status, 200, data)
+        self.assertEqual(data["user_id"], self.member_id)
+        self.assertNotIn(self.member_id, self._member_ids())
+        self.assertIn(("remove_member", self.member_id),
+                      self._audit_actions())
+
+    def test_non_owner_and_outsider_denied(self):
+        self.assertEqual(self._invite(self.member_id)[0], 201)
+        status, _ = self._delete(self.member_id, self.maint_key)
+        self.assertEqual(status, 403)
+        status, _ = self._delete(self.member_id, self.out_key)
+        self.assertEqual(status, 403)
+        # 拒绝后成员仍在。
+        self.assertIn(self.member_id, self._member_ids())
+
+    def test_global_admin_delete_without_membership(self):
+        self.assertEqual(self._invite(self.member_id)[0], 201)
+        status, data = self._delete(self.member_id, self.admin_key)
+        self.assertEqual(status, 200, data)
+        self.assertNotIn(self.member_id, self._member_ids())
+        self.assertIn(("remove_member", self.member_id),
+                      self._audit_actions())
+
+    def test_delete_non_member_and_missing_project(self):
+        status, _ = self._delete("ghost", self.owner_key)
+        self.assertEqual(status, 404)
+        status, _ = _req(self.port, "DELETE",
+                         "/projects/no-such/members/%s" % self.member_id,
+                         key=self.owner_key)
+        self.assertEqual(status, 404)
+
+    def test_cannot_remove_only_owner(self):
+        status, data = self._delete(self.owner_id, self.owner_key)
+        self.assertEqual(status, 409, data)
+        status, _ = self._delete(self.owner_id, self.admin_key)
+        self.assertEqual(status, 409)
+        # 升第二位 owner 后，删原 owner 放行。
+        self.assertEqual(self._invite(self.member_id)[0], 201)
+        status, _ = _req(self.port, "PATCH",
+                         "/projects/%s/members/%s"
+                         % (self.project_id, self.member_id),
+                         {"role": "owner"}, self.owner_key)
+        self.assertEqual(status, 200)
+        status, _ = self._delete(self.owner_id, self.maint_key)
+        # maint 不是 owner 也不是全局管理员：仍 403（守卫先行）。
+        self.assertEqual(status, 403)
+        status, data = self._delete(self.owner_id, self.member_key)
+        self.assertEqual(status, 200, data)
+        self.assertNotIn(self.owner_id, self._member_ids(self.member_key))

cloud/tests/test_p1_security_worker.py

@@ -449,7 +449,8 @@ class P1WorkerTestCase(unittest.TestCase):
         self.assertEqual(_status(self.db_path, "T1"), "failed")
         self.assertEqual(self.calls, [])
 
-    # 15. PR 非 local：显式 needs_local_worker，任务与排队都不动
+    # 15. PR 非 local：显式 needs_local_worker，行状态持久化，
+    # 任务本身不动，标记过的行仍可被 local worker 认领
     def test_pr_non_local_needs_local_worker(self):
         conn = sqlite3.connect(self.db_path)
         try:
@@ -470,9 +471,14 @@ class P1WorkerTestCase(unittest.TestCase):
             row = conn.execute(
                 "SELECT status FROM pr_jobs WHERE id = ?",
                 (job_id,)).fetchone()
+            self.assertEqual(row[0], "needs_local_worker")
+            # 标记过的行仍可被 local worker 认领，不会被搁置。
+            from cloud.pr_exec import claim_pr_job
+            claimed = claim_pr_job(conn, "test-host")
+            self.assertIsNotNone(claimed)
+            self.assertEqual(claimed["id"], job_id)
         finally:
             conn.close()
-        self.assertEqual(row[0], "queued")
 
     # 16. 多文件签名 + 签名缺失/已变时认领 fail-closed
     def test_code_signature_multi_file_and_claim_refusal(self):

cloud/tests/test_pr_jobs.py

@@ -308,7 +308,7 @@ class WorkerPrJobsTestCase(unittest.TestCase):
         finally:
             conn.close()
         # P1：Docker worker 不再静默返回空，改为显式 needs_local_worker，
-        # 任务保持 queued 等 local worker 认领。
+        # 行状态持久化为 needs_local_worker、仍可被 local worker 认领。
         jobs = worker._run_pr_jobs()
         self.assertEqual(len(jobs), 1)
         self.assertEqual(jobs[0]["status"], "needs_local_worker")

cloud/tests/test_pr_needs_local.py

@@ -0,0 +1,116 @@
+"""PR needs_local_worker 持久化单测（残留批次 A 任务 2）。
+
+- 非 local worker 扫到 queued 的 pr 任务后，行状态持久化为
+  needs_local_worker（重读仍在），且返回条目状态一致；
+- 标记过的行仍可被 local worker 认领（claim 后 running，可
+  正常 finish），不会被搁置；
+- find_open_pr_job 仍把标记过的行算未完成（/pr 不重排）；
+- 标记是 CAS：已 running 的行不被覆盖；重复扫不重复记。
+
+测试用的 key、主机名、任务 id 全是假值；不联网、不碰生产。
+"""
+
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.db import connect as _db_connect
+    from cloud.pr_exec import (
+        claim_pr_job, enqueue_pr_job, find_open_pr_job,
+        mark_pr_job_needs_local_worker)
+    from cloud.sandbox import DockerExecutor
+    from cloud.worker import Worker
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.db import connect as _db_connect
+    from cloud.pr_exec import (
+        claim_pr_job, enqueue_pr_job, find_open_pr_job,
+        mark_pr_job_needs_local_worker)
+    from cloud.sandbox import DockerExecutor
+    from cloud.worker import Worker
+
+
+class PrNeedsLocalTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+
+    def _enqueue_t1(self, host="host-a"):
+        conn = _db_connect(self.db_path)
+        try:
+            enqueue_pr_job(conn, "t1", "p1", host)
+        finally:
+            conn.close()
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _row(self, task_id):
+        conn = _db_connect(self.db_path)
+        try:
+            return conn.execute(
+                "SELECT id, status, error FROM pr_jobs"
+                " WHERE task_id = ?", (task_id,)).fetchone()
+        finally:
+            conn.close()
+
+    def test_docker_peek_persists_mark(self):
+        worker = Worker(self.db_path, "/nonexistent/host.py",
+                        env_key="TEST_KEY",
+                        executor=DockerExecutor("/fake/host.py"))
+        conn = _db_connect(self.db_path)
+        try:
+            enqueue_pr_job(conn, "t1", "p1", worker.host_name)
+        finally:
+            conn.close()
+        jobs = worker._run_pr_jobs()
+        self.assertEqual(len(jobs), 1)
+        self.assertEqual(jobs[0]["status"], "needs_local_worker")
+        self.assertEqual(jobs[0]["task_id"], "t1")
+        # 重读：行状态已持久化，不只是返回值里有。
+        row = self._row("t1")
+        self.assertEqual(row[1], "needs_local_worker")
+        self.assertIn("local worker", row[2])
+        # 再扫一次：已标记的行不再是 queued，不重复返回、不重复记。
+        self.assertEqual(worker._run_pr_jobs(), [])
+
+    def test_marked_row_still_claimable_by_local(self):
+        self._enqueue_t1()
+        conn = _db_connect(self.db_path)
+        try:
+            job_id = conn.execute(
+                "SELECT id FROM pr_jobs WHERE task_id = 't1'"
+            ).fetchone()[0]
+            self.assertTrue(mark_pr_job_needs_local_worker(
+                conn, job_id, "needs local worker"))
+            # 标记后仍算未完成：/pr 去重不受影响。
+            self.assertIsNotNone(find_open_pr_job(conn, "t1"))
+            # local worker 认领标记过的行。
+            claimed = claim_pr_job(conn, "host-a")
+            self.assertIsNotNone(claimed)
+            self.assertEqual(claimed["task_id"], "t1")
+            row = conn.execute(
+                "SELECT status FROM pr_jobs WHERE task_id = 't1'"
+            ).fetchone()
+            self.assertEqual(row[0], "running")
+        finally:
+            conn.close()
+
+    def test_mark_is_cas_on_queued_only(self):
+        self._enqueue_t1()
+        conn = _db_connect(self.db_path)
+        try:
+            claimed = claim_pr_job(conn, "host-a")
+            self.assertIsNotNone(claimed)
+            # 已 running：标记必须失败、不覆盖状态。
+            self.assertFalse(mark_pr_job_needs_local_worker(
+                conn, claimed["id"], "late mark"))
+            row = conn.execute(
+                "SELECT status FROM pr_jobs WHERE task_id = 't1'"
+            ).fetchone()
+            self.assertEqual(row[0], "running")
+        finally:
+            conn.close()

cloud/tests/test_push_toctou.py

@@ -0,0 +1,172 @@
+"""推送 DNS TOCTOU 单测（残留批次 A 任务 3）。
+
+旧版：validate_push_target 解析一次，urlopen 建连时又解析
+一次；两次解析结果不一致（DNS 重绑定）时可能拨向未验证
+的内网 IP。新版：建连时解析一次并判段、只拨判过的 IP，
+Host 头与 TLS SNI 仍为原域名。
+
+- 两次解析不一致（先公网、后内网）时：发送失败，且一次
+  都没有拨向内网 IP；
+- 混合解析结果（先内网、后公网）时：只拨公网 IP；
+- HTTP 真服务器：域名经 resolver 钉到 127.0.0.1，服务器
+  收到的 Host 仍为原域名；
+- HTTPS 连接类：拨号地址为判过的 IP，wrap_socket 的
+  server_hostname 仍为原域名（SNI 不变）。
+
+测试用的 IP、域名、key 全是假值；除本地回环服务器外不联网。
+"""
+
+import http.client
+import http.server
+import os
+import sys
+import threading
+import unittest
+from pathlib import Path
+from unittest import mock
+
+try:
+    from cloud.push import (
+        _checked_connection_class, _post_json, send_webhook)
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.push import (
+        _checked_connection_class, _post_json, send_webhook)
+
+PUBLIC_IP = "8.8.8.8"
+PRIVATE_IP = "169.254.169.254"
+
+
+class _FlipFlopResolver:
+    """第一次返回公网 IP，之后返回内网 IP：模拟重绑定。
+
+    旧版两次解析恰好踩中这个窗口；新版第二次解析判段后
+    直接拒绝，绝不拨号。
+    """
+
+    def __init__(self):
+        self.calls = 0
+
+    def __call__(self, host):
+        self.calls += 1
+        return [PUBLIC_IP] if self.calls == 1 else [PRIVATE_IP]
+
+
+class _DialRecorder:
+    """记录拨号地址后直接失败：测试只断言拨向，不跑完 HTTP。"""
+
+    def __init__(self):
+        self.calls = []
+
+    def __call__(self, address, timeout=None, source_address=None):
+        self.calls.append(address)
+        raise OSError("dial stopped by test")
+
+
+class _DummySocket:
+    def setsockopt(self, *args):
+        pass
+
+
+class _RecordingHandler(http.server.BaseHTTPRequestHandler):
+    requests = []
+
+    def do_POST(self):
+        length = int(self.headers.get("Content-Length", 0))
+        body = self.rfile.read(length) if length else b""
+        _RecordingHandler.requests.append({
+            "host": self.headers.get("Host"),
+            "body": body,
+        })
+        self.send_response(200)
+        self.send_header("Content-Length", "2")
+        self.end_headers()
+        self.wfile.write(b"{}")
+
+    def log_message(self, *args):
+        pass
+
+
+class PushToctouTestCase(unittest.TestCase):
+    def setUp(self):
+        self._old_allow = os.environ.get("CODEX_PUSH_ALLOW_PRIVATE_TARGETS")
+        os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None)
+        _RecordingHandler.requests = []
+
+    def tearDown(self):
+        if self._old_allow is None:
+            os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None)
+        else:
+            os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = self._old_allow
+
+    def _message(self, target, resolver=None):
+        msg = {"event_type": "completed", "project_id": "proj-a",
+               "payload": {"task_id": "t-1"}, "target": target}
+        if resolver is not None:
+            msg["_resolver"] = resolver
+        return msg
+
+    def test_rebinding_second_resolution_never_dialed(self):
+        resolver = _FlipFlopResolver()
+        dials = _DialRecorder()
+        with mock.patch("socket.create_connection", dials):
+            with self.assertRaises(RuntimeError):
+                send_webhook(self._message(
+                    "http://hook.example.invalid/hook", resolver))
+        # 两次解析确实都发生了（一次预检、一次建连），但建连侧
+        # 判段后拒绝：内网 IP 一次都没被拨。
+        self.assertEqual(resolver.calls, 2)
+        self.assertEqual(dials.calls, [])
+
+    def test_mixed_resolution_dials_only_public(self):
+        dials = _DialRecorder()
+        with mock.patch("socket.create_connection", dials):
+            ok, _detail = _post_json(
+                "http://hook.example.invalid/hook", {"a": 1},
+                resolver=lambda host: [PRIVATE_IP, PUBLIC_IP])
+        self.assertFalse(ok)
+        self.assertEqual(dials.calls, [(PUBLIC_IP, 80)])
+
+    def test_http_pinned_host_header_preserved(self):
+        os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1"
+        server = http.server.HTTPServer(("127.0.0.1", 0),
+                                        _RecordingHandler)
+        port = server.server_address[1]
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            send_webhook(self._message(
+                "http://webhook.test.invalid:%d/hook" % port,
+                resolver=lambda host: ["127.0.0.1"]))
+        finally:
+            server.shutdown()
+            server.server_close()
+            thread.join(timeout=5)
+        self.assertEqual(len(_RecordingHandler.requests), 1)
+        host_header = _RecordingHandler.requests[0]["host"]
+        # 拨的是 127.0.0.1，但 Host 仍为原域名（虚拟主机依赖它）。
+        self.assertEqual(host_header.split(":")[0],
+                         "webhook.test.invalid")
+
+    def test_https_sni_preserved_with_pinned_ip(self):
+        wrap_calls = []
+
+        class FakeContext:
+            def wrap_socket(self, sock, server_hostname=None):
+                wrap_calls.append(server_hostname)
+                return sock
+
+        cls = _checked_connection_class(http.client.HTTPSConnection,
+                                        lambda host: [PUBLIC_IP])
+        conn = cls("example.com", 443)
+        conn._context = FakeContext()
+        dials = []
+
+        def fake_create(address, timeout=None, source_address=None):
+            dials.append(address)
+            return _DummySocket()
+
+        with mock.patch("socket.create_connection", fake_create):
+            conn.connect()
+        self.assertEqual(dials, [(PUBLIC_IP, 443)])
+        self.assertEqual(wrap_calls, ["example.com"])

cloud/worker.py

@@ -153,7 +153,8 @@ from cloud.sub2api import gateway_name as _gateway_name
 from cloud.sub2api import gateway_origin as _gateway_origin
 from cloud.sub2api import list_gateway_models as _list_gateway_models
 from cloud.pr import PrSkipped, PrStepError
-from cloud.pr_exec import (claim_pr_job, execute_task_pr, finish_pr_job)
+from cloud.pr_exec import (claim_pr_job, execute_task_pr, finish_pr_job,
+                           mark_pr_job_needs_local_worker)
 from cloud.queue import budget_exceeded, ready_tasks
 from cloud.repos import RepoError, checkout_dir, resolve_repo
 from cloud.sandbox import CloudExecutor, DockerExecutor, LocalExecutor
@@ -3547,19 +3548,15 @@ class Worker:
         P1 非 local 通道：旧版非 local worker 直接返回 []，
         排队的 pr 任务看起来像“没人处理”且无任何痕迹——
         静默未完成。现在改为显式 needs_local_worker：只读
-         peek 本主机 queued 的 pr 任务，逐条记审计 + 日志并
-        返回 needs_local_worker 条目，任务本身保持 merged、
-        pr 任务保持 queued（留给 local worker 认领），绝不
-        把它标 failed（failed 会让调用方以为 PR 已失败，且
-        任务仍可重调 /pr 的语义会被混淆）。为什么不直接把
-        pr_jobs 行写成 needs_local_worker 状态：pr_jobs 的认
-        领只认 queued（claim_pr_job 在 pr_exec.py，本路不改
-        该文件），写成新状态后 local worker 也认领不到，任务
-        会被永久搁置；真要落库状态，需 pr_exec.py 的认领同
-        步改成含 needs_local_worker 或 local 重排队——该改动
-        留给协调员（见汇报残留），本路不伪造一个“写了状态
-        却没人能接”的通道。签名无效时同样不碰任务（返回 []，
-        只记日志），fail-closed。
+        peek 本主机 queued 的 pr 任务，逐条把行状态持久化
+        为 needs_local_worker（CAS，只改仍是 queued 的行，
+        已被 local worker 认领的不覆盖），并记审计 + 日志
+        返回 needs_local_worker 条目。任务本身保持 merged，
+        pr 任务保持可被 local worker 认领（claim_pr_job
+        同时认 queued 与 needs_local_worker），绝不把它标
+        failed（failed 会让调用方以为 PR 已失败，且任务仍
+        可重调 /pr 的语义会被混淆）。签名无效时同样不碰
+        任务（返回 []，只记日志），fail-closed。
         """
         if not self._code_signature_valid():
             log_event(self.logger, "worker_pr_refused_code_signature",
@@ -3588,8 +3585,23 @@ class Worker:
                         continue
                     reason = ("PR 执行需要 local worker（host-worktrees"
                               " 只在主机上可见），本 worker 为 %s，"
-                              "任务保持 queued 等 local worker 认领"
+                              "任务保持可认领等 local worker 认领"
                               % self._executor_name())
+                    marked = False
+                    try:
+                        conn = _db_connect(self.db_path)
+                        try:
+                            marked = mark_pr_job_needs_local_worker(
+                                conn, job_id, reason)
+                        finally:
+                            conn.close()
+                    except Exception as exc:
+                        log_event(self.logger, "pr_job_mark_failed",
+                                  message="pr job mark failed: %s" % exc)
+                        continue
+                    if not marked:
+                        # 已被 local worker 认领或已终态：不覆盖、不重复记。
+                        continue
                     self._audit(project_id, "pr_job_needs_local_worker",
                                 "%s: %s" % (job_id, reason))
                     log_event(self.logger, "pr_job_needs_local_worker",