Cloud dispatch pool + auto PR (opt-in, skip clearly)

Worker pool: workers table + register/heartbeat (heartbeat is
registration, no 404), GET /workers with alive (120s default),
POST /workers/heartbeat (X-API-Key, missing worker_id 400).
Tasks gain executor (local/docker/cloud, None = any worker);
worker claim filters by executor in pick_next and in the CAS
WHERE clause, CAS semantics unchanged; CloudExecutor _name
now correctly returns cloud, not local.
CloudExecutor: generic HTTP runner interface (endpoint from
explicit arg or CLOUD_RUNNER_ENDPOINT, auth header name only,
token value only from env into the request header, never into
logs/body/errors); no endpoint configured = explicit failure,
never silent downgrade. Vendor adapters (ECS/Cloud Run) are
out of scope, documented as such.
Auto PR: projects.auto_pr (0/1, create-time only, no PATCH
route exists); POST /tasks/{id}/pr (maintainer+ via
approve_merge) only for merged tasks (else 409); skip reasons
in order: auto_pr_disabled, no_worktree, no_remote, no_gh,
gh_not_authenticated — all return 200 with pr_skipped, task
stays merged, no push attempted; push/pr-create failure
returns 502 with step, task stays merged; success writes
pr_url/pr_number and audits pr_created. No default auto-push
from the worker. Remote URLs in errors are scrubbed of
userinfo; gh credentials are managed by gh itself, never
stored by the control plane.
Tests: host 27 pass, cloud 323 pass (8 skipped are
environment-dependent real-Postgres tests).

cloud/worker.py

@@ -101,6 +101,7 @@ import json
 import os
 import re
 import signal
+import socket
 import subprocess
 import sys
 import time
@@ -113,8 +114,9 @@ from cloud.logging_config import get_logger, log_event, setup_logging
 from cloud.models import ModelRouter, is_fallbackable, record_fallback
 from cloud.queue import budget_exceeded, ready_tasks
 from cloud.repos import RepoError, resolve_repo
-from cloud.sandbox import DockerExecutor, LocalExecutor
+from cloud.sandbox import CloudExecutor, DockerExecutor, LocalExecutor
 from cloud.sandbox import _default_runner as _sandbox_default_runner
+from cloud.workers import heartbeat_worker, register_worker
 
 # 与 server.TERMINAL_STATUSES 对齐：这些状态的任务不再被认领或取消。
 TERMINAL_STATUSES = frozenset(
@@ -178,6 +180,11 @@ def _task_from_row(row):
         depends_on = []
     if not isinstance(depends_on, list):
         depends_on = [depends_on]
+    def _col(name, default=None):
+        try:
+            return row[name]
+        except (KeyError, IndexError, TypeError):
+            return default
     return {
         "id": row["id"],
         "project_id": row["project_id"],
@@ -190,6 +197,9 @@ def _task_from_row(row):
         "created_at": row["created_at"],
         "updated_at": row["updated_at"],
         "last_heartbeat": row["last_heartbeat"],
+        # 老库行缺 executor 列时给 None（表示任意 worker 可认领），
+        # 不炸；列本身由 _ensure_result_columns 在写回时补。
+        "executor": _col("executor"),
     }
 
 
@@ -208,7 +218,8 @@ class Worker:
                  cost_store=None, process_factory=None,
                  task_timeout=1800, max_retries=1,
                  retry_backoff_seconds=5.0, repos_dir="./repos",
-                 repos_runner=None, heartbeat_file=None):
+                 repos_runner=None, heartbeat_file=None,
+                 worker_id=None, capacity=1):
         self.db_path = str(db_path)
         self.host_py_path = host_py_path
         self.config_path = config_path
@@ -259,6 +270,39 @@ class Worker:
         self.heartbeat_file = heartbeat_file or \
             os.environ.get("CLOUD_WORKER_HEARTBEAT_FILE") or \
             "./worker-heartbeat.json"
+        # 多 worker 池身份：显式 --worker-id 优先（生产部署应配稳定
+        # id）；未给时用“主机名-进程号”，重启后进程号会变，旧行
+        # 自然变 stale，不要用随机 uuid 让注册表无法对应同一台机器。
+        self.worker_id = worker_id or "%s-%d" % (
+            socket.gethostname(), os.getpid())
+        try:
+            self.capacity = int(capacity)
+        except (TypeError, ValueError):
+            self.capacity = 1
+
+    def register(self):
+        """向控制面注册本 worker（写 workers 表，失败只记日志）。
+
+        注册与心跳共用同一张表：心跳就是带 last_seen 刷新的注册，
+        未注册的 worker 发心跳时自动注册，不 404。
+        """
+        try:
+            register_worker(
+                self.db_path, self.worker_id,
+                executor=self._executor_name(), capacity=self.capacity)
+        except Exception as exc:
+            log_event(self.logger, "worker_register_failed",
+                      message="worker register failed: %s" % exc)
+
+    def _registry_heartbeat(self):
+        """向控制面写注册表心跳（与文件心跳并行，失败只记日志）。"""
+        try:
+            heartbeat_worker(
+                self.db_path, self.worker_id,
+                executor=self._executor_name(), capacity=self.capacity)
+        except Exception as exc:
+            log_event(self.logger, "worker_registry_heartbeat_failed",
+                      message="worker registry heartbeat failed: %s" % exc)
 
     def _write_heartbeat(self):
         """写一次运维心跳：写失败只记日志，绝不炸 worker。
@@ -312,17 +356,34 @@ class Worker:
             conn.close()
         return _task_from_row(row) if row is not None else None
 
-    def _claim(self, task_id):
-        """CAS 认领：仅当任务仍为 queued 时标 running 并记心跳。"""
+    def _claim(self, task_id, executor=None):
+        """CAS 认领：仅当任务仍为 queued 时标 running 并记心跳。
+
+        executor 过滤（多 worker 池）：executor 给定时，WHERE 里
+        再加 (executor IS NULL OR executor = ?)，即只认领未指定
+        执行器、或指定为本 worker 执行器类型的任务；CAS 语义不变，
+        仍是“仅当 queued 时标 running”，两个同池 worker 抢同一
+        任务时只有一个能拿到 rowcount == 1。
+        """
         now = time.time()
         conn = _db_connect(self.db_path)
         try:
-            cur = conn.execute(
-                "UPDATE tasks SET status = 'running',"
-                " last_heartbeat = ?, updated_at = ?"
-                " WHERE id = ? AND status = 'queued'",
-                (now, now, task_id),
-            )
+            self._ensure_result_columns(conn)
+            if executor is None:
+                cur = conn.execute(
+                    "UPDATE tasks SET status = 'running',"
+                    " last_heartbeat = ?, updated_at = ?"
+                    " WHERE id = ? AND status = 'queued'",
+                    (now, now, task_id),
+                )
+            else:
+                cur = conn.execute(
+                    "UPDATE tasks SET status = 'running',"
+                    " last_heartbeat = ?, updated_at = ?"
+                    " WHERE id = ? AND status = 'queued'"
+                    " AND (executor IS NULL OR executor = ?)",
+                    (now, now, task_id, executor),
+                )
             conn.commit()
             return cur.rowcount == 1
         finally:
@@ -355,7 +416,9 @@ class Worker:
         for name, coltype in {
                 "result_text": "TEXT", "error_text": "TEXT",
                 "actual_model": "TEXT", "started_at": "REAL",
-                "finished_at": "REAL", "child_pid": "INTEGER"}.items():
+                "finished_at": "REAL", "child_pid": "INTEGER",
+                "executor": "TEXT", "pr_url": "TEXT",
+                "pr_number": "INTEGER"}.items():
             if name not in cols:
                 try:
                     conn.execute(
@@ -484,6 +547,12 @@ class Worker:
             ready = ready_tasks(tasks, completed_ids)
         except ValueError:
             return None
+        # 多 worker 池过滤：只取未指定执行器、或指定为本 worker
+        # 执行器类型的任务；_claim 里还有一道 WHERE 过滤兜底，
+        # pick_next 这里是第一道，避免 worker 老去抢别人的任务。
+        mine = self._executor_name()
+        ready = [t for t in ready
+                 if t.get("executor") in (None, "", mine)]
         return ready[0] if ready else None
 
     # -- 取消（server cancel 语义，见模块 docstring）-------------------
@@ -572,9 +641,13 @@ class Worker:
 
     # -- 缺 key 门 ------------------------------------------------------
     def _executor_name(self):
-        """执行器名：DockerExecutor -> docker，其它 -> local。"""
+        """执行器名：DockerExecutor -> docker，CloudExecutor -> cloud，
+        其它 -> local。cloud 绝不能回落成 local，否则 cloud worker
+        会认领 local 池的任务，池过滤就失效了。"""
         if isinstance(self.executor, DockerExecutor):
             return "docker"
+        if isinstance(self.executor, CloudExecutor):
+            return "cloud"
         return "local"
     def key_available(self):
         """只查存在性：名在环境里且值非空即 True；值本身绝不返回。"""
@@ -800,11 +873,15 @@ class Worker:
         if task is None:
             return {"dispatched": False, "reason": "no_ready_task"}
         task_id = task["id"]
-        if not self._claim(task_id):
+        if not self._claim(task_id, executor=self._executor_name()):
             return {"dispatched": False, "reason": "already_claimed",
                     "task_id": task_id}
         task = self.get_task(task_id) or task
-        self._audit(task["project_id"], "worker_claim", task_id)
+        self._audit(task["project_id"], "worker_claim",
+                    "%s executor=%s" % (task_id, self._executor_name()))
+        # log_event 只接受 task_id/project_id/status/message 四个
+        # 字段，executor 只进审计 detail，不进日志调用（否则
+        # TypeError 会炸掉整个 run_one）。
         log_event(self.logger, "worker_claim", task_id=task_id,
                   project_id=task["project_id"], status="running")
         started_at = time.time()
@@ -982,6 +1059,9 @@ class Worker:
         # 运维心跳：启动时先写一次，之后每轮结束再写，server 的
         # /health 据此算 worker_ok（见 server.py 的健康检查）。
         self._write_heartbeat()
+        # 多 worker 池注册：启动时注册一次，之后每轮结束写注册表
+        # 心跳（见本轮结束处）；注册失败只记日志，不炸循环。
+        self.register()
         started = time.time()
         all_results = []
         try:
@@ -1001,8 +1081,9 @@ class Worker:
                     results = []
                 all_results.extend(results)
                 # 每轮结束写心跳：哪怕本轮一个任务都没派，worker 活着
-                # 这件事本身也要被 server 看到。
+                # 这件事本身也要被 server 看到（文件心跳 + 注册表心跳）。
                 self._write_heartbeat()
+                self._registry_heartbeat()
                 # 按 0.2 秒切片睡，期间可响应 stop_event/_loop_stop。
                 deadline = time.time() + interval
                 while time.time() < deadline:
@@ -1032,7 +1113,11 @@ class Worker:
 def _build_cli_executor(args, host_py):
     """CLI 的执行器构造：local 时返回 None（Worker 默认 LocalExecutor），
     docker 时返回 DockerExecutor（env_key 名用 Worker 的解析规则，
-    值本身绝不从这里传——DockerExecutor 只查存在性，-e 只传名）。"""
+    值本身绝不从这里传——DockerExecutor 只查存在性，-e 只传名），
+    cloud 时返回 CloudExecutor（通用 HTTP runner，未配 endpoint
+    时执行即明确失败，绝不降级）。"""
+    if getattr(args, "executor", "local") == "cloud":
+        return CloudExecutor()
     if getattr(args, "executor", "local") != "docker":
         return None
     return DockerExecutor(
@@ -1075,10 +1160,17 @@ def main(argv=None):
     parser.add_argument("--max-turns", type=int, default=None)
     parser.add_argument("--max-tasks", type=int, default=None)
     parser.add_argument("--executor", default="local",
-                        choices=["local", "docker"],
+                        choices=["local", "docker", "cloud"],
                         help="执行器：local 调 host.py 子进程；docker"
                              " 调 DockerExecutor（无 docker 时任务"
-                             "明确失败，不静默降级）")
+                             "明确失败，不静默降级）；cloud 调通用"
+                             " HTTP runner（需 CLOUD_RUNNER_ENDPOINT）")
+    parser.add_argument("--worker-id", default=None,
+                        help="多 worker 池身份（默认：主机名-进程号；"
+                             "生产部署应配稳定 id，否则重启后旧行变 stale）")
+    parser.add_argument("--worker-capacity", type=int, default=1,
+                        help="worker 并发容量声明（注册表里展示用，"
+                             "本轮 run_loop 仍是串行一轮一任务）")
     parser.add_argument("--docker-image", default="codex-app-server:latest")
     parser.add_argument("--docker-cpus", default="2")
     parser.add_argument("--docker-memory", default="2g")
@@ -1119,7 +1211,9 @@ def main(argv=None):
                     max_retries=args.max_retries,
                     retry_backoff_seconds=args.retry_backoff_seconds,
                     repos_dir=args.repos_dir,
-                    heartbeat_file=args.heartbeat_file)
+                    heartbeat_file=args.heartbeat_file,
+                    worker_id=args.worker_id,
+                    capacity=args.worker_capacity)
     # CLI 启动清理兜底（常驻与单次都做）：清掉上一轮残留容器。
     worker.cleanup_stale_containers()
     # CLI 起 worker 前先建表/迁移：全新库直接起循环不应因缺表而炸。