"""云端 worker：把控制面的 queued 任务接通到 host.py 执行。 零第三方依赖，只用 Python 标准库；不改 host.py 行为、不动 ~/.codex 配置、不碰 wechat-bridge。 流程（与 server.py 的任务状态机对齐）： queued -> running -> completed / failed / interrupted 1. 认领：pick_next() 用 queue.ready_tasks() 按依赖拓扑 + 优先级 排序，跳过依赖未完成（未 completed/merged）的任务；再过 queue.budget_exceeded() 预算门，超限不派发、任务保持 queued。 claim() 用 `WHERE status='queued'` 的 CAS 更新把任务标 running 并记 last_heartbeat / updated_at，避免两个 worker 重复认领。 2. 缺 key 门（执行前必查）：只检查配置的 env_key 名在进程环境 里是否存在且非空，绝不读出、打印、写进文件或日志 key 的值。 缺 key 时任务直接标 failed，错误固定为 `Missing environment variable: `，与之前 real-nightly-tests automation 的失败模式一致，绝不静默假装跑了。 env_key 名的来源：显式参数 > codex 配置文件里的 env_key 行 （只读名字）> 默认 XCM_API_KEY（中转站 relay 的约定名）。 3. 执行：经 LocalExecutor 调 host.py 真实子命令。dispatch （建 host 任务并跑一轮 turn）是首次派发；dispatch 报 “任务已存在”时改用 run（在已有 host 任务上再跑一轮）续跑。 host 任务名由 cloud 任务 id 派生（见 sandbox._host_name）， 同一任务每次算出同一名字，续跑才能找到同一个 host 任务。 4. 写回：server 的 tasks 表没有 result/error 列，所以执行结果 的持久化分三处，各复用已有结构、不另起字段名： - tasks 表：最终状态 + updated_at（取消时为 interrupted）； - cost.py CostStore：每轮实际执行记一条 source="estimated" 的成本记录（host 协议不暴露真实 token 数，估算绝不冒充 actual；total_tokens 由 CostStore.record 按 input+output 算）； - models.py：可 fallback 的失败（404/429/5xx/timeout，401/400 不算）。默认（未配 --fallback-models）时 worker 只记录 fallback 事件（record_fallback() 记 model_fallback 事件， fallback_count 写进成本记录），不自动换模型重试——自动重试 等于未经批准烧 relay 额度。显式配了 --fallback-models 时， 第一次失败且可 fallback 就真的换列表里的第一个模型再跑一次 （最多 1 次，同一任务、同一 worktree/分支，dispatch 报 “任务已存在”时走现有的 run 续跑路径，模型经 host.py 的 顶层 --model 参数传过去，不改 host.py 签名）；第二次的结果 就是最终结果，actual_model 写成最后跑的模型，fallback_count=1。 审计：认领/完成/失败/取消/缺 key 各记一条 AuthStore 审计 （user_id="system"，与 server.sweep_stale_tasks 的做法一致）， 审计失败不影响任务状态（best-effort）。 取消：worker 的 cancel() 走 server 的 cancel 语义：把 queued/running 任务直接标 interrupted（已终态的任务不动）， 且正在跑的子进程会真杀——子进程以进程组启动（start_new_session）， 取消时先 SIGTERM 给整个进程组、等 2 秒、仍活着再 SIGKILL， 杀完后任务保持 interrupted，执行结果不覆盖它。run_one() 在 执行返回后会先查任务是否已被取消，已被取消的一律保持 interrupted，不用执行结果覆盖它。 任务级超时与失败重试（生产级，见 Worker.__init__ 参数）： - task_timeout（默认 1800 秒，CLI --task-timeout 可调）是 worker 级的每任务截止时间，与下面两个参数不要混淆： --docker-timeout 是执行器级的单次 docker run 超时（Docker Executor 自身的 timeout），--max-seconds 是 run_loop 的 整体墙钟上限（不是每任务）。三者互不覆盖：Docker 路径下 worker 取 min(task_timeout, docker_timeout)（docker_timeout 未设时取 task_timeout）作为本次 docker run 的超时，超时后 真杀容器、任务置 failed（超时是失败，不是取消，绝不标 interrupted）；Local popen 路径下 worker 在轮询循环里到 截止时间就杀进程组、同样置 failed；Local 同步路径（假 runner / 无超时的 subprocess.run）无法在中途抢占一个已阻塞 的调用，超时只能靠执行器自身的 timeout，这是诚实边界， 不要假装同步路径也能中途杀掉。 - max_retries（默认 1 次重试，CLI --max-retries 可调，0 表示 不重试）是对模型任务失败的有限重试：取消不重试；可 fallback 的失败优先走真实 fallback（见下），不再做同模型重试（fallback 本身就是模型级的重试，再同模型重试会重复烧额度）；其它失败 （含 worker 超时）重试前按 retry_backoff_seconds（默认 5 秒， 测试可注入 0）等待，重试次数记在审计（worker_retry）与 最终 error_text（[retries=N] 前缀）里，不另起新表；重试 耗尽后任务才最终 failed，成本记录与 usage_turns 只记一次 - 真实 fallback（CLI --fallback-models，默认空 = 关闭）：一次 尝试失败且可 fallback（404/429/5xx/timeout，401/400 不算， 取消不算）时，若配了 fallback 列表，就记 model_fallback 事件 与 worker_fallback 审计，然后换列表里的第一个模型再跑一次 （最多 1 次，与 max_retries 无关，max_retries=0 时 fallback 照样跑）；第二次的结果就是最终结果，actual_model 写成最后 跑的模型，fallback_count=1，两次都失败时 error_text 同时写清 两次原因；未配列表时保持老行为（只记事件、直接 failed）。 （按最终结果记，不按尝试次数重复记）。 - 容器清理兜底：Worker.cleanup_stale_containers() 列出 codex- 前缀的残留容器并逐个 kill+rm，失败只记日志不炸 worker；run_loop 启动时与 CLI 启动时各调一次（run_all / __init__ 里不调，避免单测构造 Worker 时多余 I/O）。 可注入：runner（假 CLI，见 sandbox.LocalExecutor）与 env 映射 均可注入，单测不跑真实模型回合、不联网、不登录任何服务。 执行器选择：Worker 默认用 LocalExecutor；传入 DockerExecutor （或 CLI --executor docker）时，派发走容器内 host.py，取消时 按任务派生的容器名 docker kill，不只改数据库状态；本机无 docker 时 DockerExecutor 在调 runner 前就明确报错，任务标 failed 且 error_text 含 docker 缺失，绝不静默降级为本地执行。 项目仓库：run_one 派发前查任务所属项目的 repo_url（projects 表），非空时经 cloud.repos.resolve_repo 在 repos_dir（CLI --repos-dir，默认 ./repos）下 clone/fetch 该项目仓库，设 task["repo"] 再交给执行器；Local 与 Docker 执行器都已读 task["repo"]，同一条路径。repo_url 为空时走原有默认 repo； 解析失败时任务直接 failed，错误文本写清是仓库解析失败。 Docker 下 repos_dir 必须放在 extra-mounts 已挂载的目录下， 否则容器内 repo 路径不存在（见 cloud/README.md）。 推送到远端不在本轮范围：任务在本地检出上执行、本地 worktree 合并仍按 host.py 现有流程，git push 需要仓库写凭证，控制面 不存、不推。 """ import argparse import json import os import re import signal import socket import subprocess import sys import threading import time from types import SimpleNamespace from cloud.auth import AuthStore from cloud.cost import CostStore, estimate_tokens from cloud.db import connect as _db_connect 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 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( {"completed", "failed", "interrupted", "merged", "merge-failed"}) COMPLETED_STATUSES = frozenset({"completed", "merged"}) DEFAULT_ENV_KEY = "XCM_API_KEY" DEFAULT_MODEL = "gpt-6.1-sol" # host.py dispatch 的原话是 "task 已存在: ，先换名或 discard 它" #（英文 task + 已存在），取共有子串做判断，不依赖标点。 _ALREADY_EXISTS_MARKER = "已存在" def _missing_key_error(env_key): """缺 key 的标准错误：只含 env_key 名，与 relay 侧的报错一致。""" return "Missing environment variable: %s" % env_key def resolve_env_key(env_key=None, codex_config_path=None): """确定模型 key 对应的环境变量名（只返回名字，从不返回值）。 显式参数优先；其次读 codex 配置文件（如 relay 的 config.toml） 里的 `env_key = "..."` 行，只取名字；文件不存在或无此行时 用默认名。绝不读 ~/.codex/config.toml 之外的默认位置—— 调用方要读哪个配置文件，必须显式传 codex_config_path。 """ if env_key: return env_key if codex_config_path: try: with open(codex_config_path, encoding="utf-8") as f: text = f.read() except (OSError, UnicodeDecodeError): text = "" match = re.search( r'^\s*env_key\s*=\s*"([^"]+)"', text, re.MULTILINE) if match: return match.group(1) return DEFAULT_ENV_KEY def _resolve_model(model=None, config_path=None): """模型名：显式参数 > host 配置文件里的 model > host 默认值。""" if model: return model if config_path: try: with open(config_path, encoding="utf-8") as f: cfg = json.load(f) except (OSError, ValueError): cfg = {} if isinstance(cfg, dict) and cfg.get("model"): return cfg["model"] return DEFAULT_MODEL def _task_from_row(row): depends_on = row["depends_on"] try: depends_on = json.loads(depends_on) if depends_on else [] except (TypeError, ValueError): 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"], "prompt": row["prompt"], "status": row["status"], "idempotency_key": row["idempotency_key"], "depends_on": depends_on, "priority": row["priority"], "created_by": row["created_by"], "created_at": row["created_at"], "updated_at": row["updated_at"], "last_heartbeat": row["last_heartbeat"], # 老库行缺 executor 列时给 None（表示任意 worker 可认领）， # 不炸；列本身由 _ensure_result_columns 在写回时补。 "executor": _col("executor"), # goal 字段同款 _col 容错：老库行与 load_tasks 的显式列清单 # 里没有时给 None（= 普通任务），不炸；run_one 认领后 # get_task 重读 SELECT *，goal 任务一定能拿到完整字段。 "goal": _col("goal"), "verify_command": _col("verify_command"), "goal_max_rounds": _col("goal_max_rounds"), } class Worker: """单机 worker：一轮认领一个 queued 任务并执行到底。 前置条件：db_path 须已由 cloud.server.init_db() 建好 tasks 表； 本模块不自建 server 的表（成本表与审计表由 CostStore/AuthStore 在首次使用时自建，与 server 的做法一致）。 """ def __init__(self, db_path, host_py_path, config_path=None, repo=None, runner=None, env=None, env_key=None, codex_config_path=None, model=None, router=None, max_turns=None, max_cost_usd=None, executor=None, 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, worker_id=None, capacity=1, fallback_models=None, tenant_isolation=False, tenants_root=None, tenant_template_home=None, tenant_template_repo=None, egress_proxy_port=0, claim_executor_only=False): self.db_path = str(db_path) self.host_py_path = host_py_path self.config_path = config_path self.repo = repo # env 只用于 key 存在性检查：默认接进程环境，单测可注入 # 一个普通 dict；worker 永远不把 env 里的值写出。 self._env = env if env is not None else os.environ self.env_key = resolve_env_key(env_key, codex_config_path) self.model = _resolve_model(model, config_path) if router is not None and not isinstance(router, ModelRouter): raise ValueError("router 必须是 cloud.models.ModelRouter") self.router = router # 真实 fallback 换模型列表：默认空 = 关闭，只记事件不换模型 # （老行为，生产 worker 未配该参数时绝不突然多花模型钱）。 # 归一化：去空白、去重保序、去掉与主模型相同的项（换成同一 # 个模型不叫 fallback，只会白烧一次额度）；fallback 总次数 # 最多 1 次，所以运行期只用列表的第一个元素，后面的只是 # 备用声明，不会被依次试下去。 self.fallback_models = [] for _m in (fallback_models or []): if not isinstance(_m, str): continue _m = _m.strip() if _m and _m != self.model and _m not in self.fallback_models: self.fallback_models.append(_m) self.max_turns = max_turns self.max_cost_usd = max_cost_usd self.usage_turns = 0 self.usage_cost_usd = 0.0 if executor is not None: self.executor = executor else: self.executor = LocalExecutor( host_py_path, config_path=config_path, runner=runner, repo=repo) self._cost_store = cost_store self._cancel_requested = set() # 真取消用：process_factory 可注入（单测用它返回一个假长跑 # 子进程，真实执行时走 LocalExecutor.popen_task 进程组启动）。 self.process_factory = process_factory # 并发运行状态：task_id -> {"proc": Popen|None, # "container": 容器名|None}，受 _running_lock 保护。 # 旧版只有单任务三个字段（_running_proc 等），capacity=1 # 时够用；capacity>1 后多个任务同时跑，单字段会被后 # 一个任务覆盖，导致 cancel 杀错、退出时漏杀。所有读写 # 都走下面的 _running_set/_running_clear/_running_snapshot。 self._running = {} self._running_lock = threading.Lock() # DB 写锁：cost 记录共用一条连接（CostStore._conn）， # usage 计数也是共享变量；模型回合本身在子进程/容器里 # 跑不受这把锁影响，锁只覆盖毫秒级的写库操作。 self._db_lock = threading.RLock() self._loop_stop = False self.logger = get_logger("cloud.worker") # 任务级超时 / 失败重试（见模块 docstring 的生产级说明）。 self.task_timeout = task_timeout self.max_retries = max_retries self.retry_backoff_seconds = retry_backoff_seconds # 项目仓库检出根目录：CLI --repos-dir 可调（默认 ./repos）； # Docker 执行器下它必须放在 extra-mounts 已挂载的目录下， # 否则容器内 task["repo"] 路径不存在——见 cloud/README.md。 # repos_runner 可注入（单测用假 git runner），缺省走真实 git。 self.repos_dir = repos_dir self.repos_runner = repos_runner # 多租户隔离（仅 Docker 执行器有效，Local 执行器在本机 # 直接跑 host.py，同一用户下本来就没有容器边界，flag # 对它不起作用）：开启后每个项目用独立的宿主目录 # tenants_root//{work,codex-home,test-repo, # }，容器内路径保持与原来完全一致（/app/work、 # /var/lib/.../codex-home 等），只是宿主侧的挂载源按 # 项目分开——容器内看不到其它项目的任何文件。codex-home # 只播种 config.toml（relay 配置，不含任何密钥值）， # 绝不复制 thread_history（那是其它任务的 prompt 与结果）。 self.tenant_isolation = tenant_isolation self.tenants_root = tenants_root or \ "/var/lib/codex-app-server/tenants" self.tenant_template_home = tenant_template_home or \ "/var/lib/codex-app-server/codex-home" self.tenant_template_repo = tenant_template_repo or \ "/var/lib/codex-app-server/test-repo" # egress 代理端口（0 = 不启用）：启用后 Docker 任务容器 # 的 HTTP(S) 出口全部走宿主上的过滤代理（网关 IP + 该 # 端口），代理只放行白名单域名；网关查不到时任务直接 # failed，绝不无代理跑（无代理等于绕过白名单）。 self.egress_proxy_port = egress_proxy_port or 0 # 严格 executor 认领：只认领 executor 恰好等于本 worker # 类型的任务，绝不认领 executor 为 NULL 的任务。默认 # False（老行为：NULL 任务任意 worker 可认领）。宿主级 # 沙箱 worker 必须开这个，否则它会抢走本该进 Docker # 容器的 NULL 任务（那些任务在宿主上跑就丢了 Docker # 外层，违背“按任务二选一”的路由意图）。 self.claim_executor_only = bool(claim_executor_only) # 运维心跳：路径优先显式参数，其次环境变量，默认当前目录下 # 的 worker-heartbeat.json；主机路径不在代码里硬编码，主机 # 部署时用 --heartbeat-file 或环境变量传入（见 deploy/README）。 # 文件里只有 last_seen 与 pid，绝无任何密钥。 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。 先写临时文件再 os.replace，保证 server 读到的要么是旧的 完整文件、要么是新的完整文件，不会读到写一半的 JSON。 """ try: payload = {"last_seen": time.time(), "pid": os.getpid()} tmp = "%s.tmp" % self.heartbeat_file parent = os.path.dirname(os.path.abspath(self.heartbeat_file)) os.makedirs(parent, exist_ok=True) with open(tmp, "w", encoding="utf-8") as f: f.write(json.dumps(payload)) os.replace(tmp, self.heartbeat_file) except Exception as exc: log_event(self.logger, "worker_heartbeat_failed", message="worker heartbeat write failed: %s" % exc) # -- 成本存储（惰性建表，避免只认领不执行时也建表）-------------- @property def costs(self): if self._cost_store is None: self._cost_store = CostStore(self.db_path) return self._cost_store def close(self): """关闭惰性建的成本存储连接（注入的 cost_store 由调用方管）。""" if self._cost_store is not None: try: self._cost_store._conn.close() except Exception: pass # -- DB 读写 ------------------------------------------------------ def load_tasks(self): """pick_next 的输入：只取 queued/completed/merged 三类任务。 旧实现是 SELECT * 全表：prompt/result_text/error_text （后两者各可达 4000 字）全量进内存，且每轮 poll 都来一次； 任务表随时间变大后这是纯浪费。pick_next 只需要 queued 任务和已完成 id（依赖判断），running/failed/interrupted 的任务对它既不是就绪、也不是已完成，排除它们不影响结果。 列清单必须覆盖 _task_from_row 直接取的全部字段（prompt、 idempotency_key 等），缺一列就会 KeyError；result_text、 error_text 等重字段不在此列，真正需要它们的 get_task 仍是 SELECT * 单行取。 """ conn = _db_connect(self.db_path) try: rows = conn.execute( "SELECT id, project_id, prompt, status, idempotency_key," " depends_on, priority, created_by, created_at, updated_at," " last_heartbeat, executor, goal, verify_command," " goal_max_rounds FROM tasks" " WHERE status IN ('queued', 'completed', 'merged')" " ORDER BY created_at, id").fetchall() finally: conn.close() return [_task_from_row(r) for r in rows] def get_task(self, task_id): conn = _db_connect(self.db_path) try: row = conn.execute( "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone() finally: conn.close() return _task_from_row(row) if row is not None else None 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。 claim_executor_only 开启时（宿主级沙箱 worker 用）：WHERE 只加 executor = ?，NULL 任务绝不认领（否则宿主 worker 会抢走本该进 Docker 的任务）。 """ now = time.time() conn = _db_connect(self.db_path) try: 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), ) elif self.claim_executor_only: cur = conn.execute( "UPDATE tasks SET status = 'running'," " last_heartbeat = ?, updated_at = ?" " WHERE id = ? AND status = 'queued'" " AND executor = ?", (now, now, task_id, executor), ) 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: conn.close() def _set_status(self, task_id, status): now = time.time() conn = _db_connect(self.db_path) try: conn.execute( "UPDATE tasks SET status = ?, last_heartbeat = ?," " updated_at = ? WHERE id = ?", (status, now, now, task_id), ) conn.commit() finally: conn.close() def _ensure_result_columns(self, conn): """worker 侧的迁移兜底：与 server.migrate_tasks_table 同列。 worker 的库理论上已由 server.init_db 迁过；这里再查一次 是为了 worker 直连老库（测试里手工建的表）时也不炸。 """ try: cols = {row[1] for row in conn.execute( "PRAGMA table_info(tasks)").fetchall()} except Exception: return for name, coltype in { "result_text": "TEXT", "error_text": "TEXT", "actual_model": "TEXT", "started_at": "REAL", "finished_at": "REAL", "child_pid": "INTEGER", "executor": "TEXT", "pr_url": "TEXT", "pr_number": "INTEGER", "goal": "TEXT", "verify_command": "TEXT", "goal_max_rounds": "INTEGER"}.items(): if name not in cols: try: conn.execute( "ALTER TABLE tasks ADD COLUMN %s %s" % (name, coltype)) except Exception: pass def _set_status_if_running(self, task_id, status): """仅当任务仍为 running 时改状态，返回是否改成功。 并发 + 优雅排空下，后台线程可能在任务已被取消 （标 interrupted）后才跑完；此时最终写回绝不能把 interrupted 盖回 completed/failed。正常流程里任务 到最终写回时仍是 running，CAS 一定成功。 """ now = time.time() conn = _db_connect(self.db_path) try: cur = conn.execute( "UPDATE tasks SET status = ?, last_heartbeat = ?," " updated_at = ? WHERE id = ? AND status = 'running'", (status, now, now, task_id), ) conn.commit() return cur.rowcount == 1 finally: conn.close() def _write_execution_result(self, task_id, result_text=None, error_text=None, actual_model=None, started_at=None, finished_at=None, child_pid=None): """把执行结果写进 tasks 新列；老库缺列时先迁移再写。""" conn = _db_connect(self.db_path) try: self._ensure_result_columns(conn) if child_pid is None: conn.execute( "UPDATE tasks SET result_text = ?, error_text = ?," " actual_model = ?, started_at = ?, finished_at = ?," " updated_at = ? WHERE id = ?", ((result_text or "")[:4000] if result_text else None, (error_text or "")[:4000] if error_text else None, actual_model, started_at, finished_at, time.time(), task_id), ) else: conn.execute( "UPDATE tasks SET result_text = ?, error_text = ?," " actual_model = ?, started_at = ?, finished_at = ?," " child_pid = ?, updated_at = ? WHERE id = ?", ((result_text or "")[:4000] if result_text else None, (error_text or "")[:4000] if error_text else None, actual_model, started_at, finished_at, child_pid, time.time(), task_id), ) conn.commit() finally: conn.close() def _set_child_pid(self, task_id, pid): conn = _db_connect(self.db_path) try: self._ensure_result_columns(conn) conn.execute( "UPDATE tasks SET child_pid = ? WHERE id = ?", (pid, task_id), ) conn.commit() finally: conn.close() def _kill_proc_group(self, proc): """真杀：先 SIGTERM 给整个进程组，等 2 秒，仍活着再 SIGKILL。 proc 须是 start_new_session 启动的 Popen（自成进程组）， 杀组才能连同它拉起的子孙一起杀掉，不只杀父进程。 """ try: pgid = os.getpgid(proc.pid) except Exception: return try: os.killpg(pgid, signal.SIGTERM) except (ProcessLookupError, PermissionError, OSError): return deadline = time.time() + 2.0 while time.time() < deadline: try: os.killpg(pgid, 0) except (ProcessLookupError, PermissionError, OSError): return time.sleep(0.05) try: os.killpg(pgid, signal.SIGKILL) except (ProcessLookupError, PermissionError, OSError): pass def heartbeat(self, task_id): """刷新 running 任务的心跳；任务不在 running 时返回 False。""" now = time.time() conn = _db_connect(self.db_path) try: cur = conn.execute( "UPDATE tasks SET last_heartbeat = ?, updated_at = ?" " WHERE id = ? AND status = 'running'", (now, now, task_id), ) conn.commit() return cur.rowcount == 1 finally: conn.close() def _audit(self, project_id, action, detail=""): """best-effort 审计：失败不影响任务状态。""" try: store = AuthStore(self.db_path) try: store.audit("system", project_id, action, detail) finally: try: store.conn.close() except Exception: pass except Exception: pass def _notify_task_finished(self, task_id, project_id, event_type, error=None): """任务终态推送：best-effort，绝不因通知失败影响任务状态。 只发终态事件：completed/failed/goal-unmet （push.ALLOWED_PUSH_EVENTS 的子集；interrupted 不在 事件集合里，取消不通知）。无订阅时 dispatch 直接返回空 sent，不联网。同步发送、单次 10 秒 超时：慢订阅会拖住 worker 循环最多几秒，换取投递结果 可观测（失败记日志，不静默丢）；后台队列是后续优化， 不在本轮。 """ try: from cloud.push import dispatch_event result = dispatch_event( self.db_path, event_type, project_id, {"task_id": task_id, "status": event_type, "executor": self._executor_name(), "error": error}) if result.get("errors"): log_event(self.logger, "worker_notify_failed", task_id=task_id, project_id=project_id, status=event_type, message="notify errors: %s" % (result["errors"],)) except Exception as exc: # noqa: BLE001 -- 通知失败绝不炸任务 log_event(self.logger, "worker_notify_failed", task_id=task_id, project_id=project_id, status=event_type, message="notify exception: %s" % exc) # -- 认领 ---------------------------------------------------------- def _budget_exceeded(self): return budget_exceeded(self.usage_turns, self.max_turns, self.usage_cost_usd, self.max_cost_usd) def _ready_tasks(self): """当前就绪任务列表（pick_next 的批量版）。 逻辑与 pick_next 完全一致（预算门、依赖满足、执行器 过滤），只是返回全部而不是第一个。自动扩缩容用它算 队列深度：只数真正能立刻派的任务，被依赖卡住的 queued 任务不算，否则 scaler 会为永远派不出去的任务 扩容。依赖成环时返回 []（与 pick_next 的 None 一致）。 """ if self._budget_exceeded(): return [] tasks = self.load_tasks() completed_ids = {t["id"] for t in tasks if t["status"] in COMPLETED_STATUSES} try: ready = ready_tasks(tasks, completed_ids) except ValueError: return [] mine = self._executor_name() if self.claim_executor_only: return [t for t in ready if t.get("executor") == mine] return [t for t in ready if t.get("executor") in (None, "", mine)] def pick_next(self): """下一个可派任务，或 None（无就绪任务 / 预算超限 / 依赖成环）。 依赖满足的定义：依赖任务状态为 completed 或 merged。 依赖成环时 ready_tasks 抛 ValueError，worker 视为无就绪任务 （成环任务本就永远派不出去，run_all 不应因此崩溃）。 """ ready = self._ready_tasks() return ready[0] if ready else None # -- 并发运行状态 -------------------------------------------------- def _running_set(self, task_id, proc=None, container=None): with self._running_lock: entry = self._running.setdefault(task_id, {}) if proc is not None: entry["proc"] = proc if container is not None: entry["container"] = container def _running_clear(self, task_id): with self._running_lock: self._running.pop(task_id, None) def _running_snapshot(self): with self._running_lock: return {tid: dict(entry) for tid, entry in self._running.items()} def _wait_for_running(self, timeout=None): """等 _running 变空；timeout 为 None 时无限等。 返回 True 表示已空，False 表示超时仍有任务在跑。 run_loop 退出时的优雅排空用它：先等任务自然结束， 超时后调用方再杀残留并标 interrupted。 """ deadline = None if timeout is None else time.time() + timeout while True: with self._running_lock: if not self._running: return True if deadline is not None and time.time() >= deadline: return False time.sleep(0.2) def _heartbeat_running_tasks(self): """给所有正在跑的任务刷 last_heartbeat。 run_loop 的心跳线程调它：任务在跑时心跳保持新鲜， 自动陈旧回收（sweep）才不会误杀活任务；worker 真死后 心跳停了，任务超过阈值才会被别的 worker 回收。 """ for task_id in self._running_snapshot(): try: self.heartbeat(task_id) except Exception: pass # -- 取消（server cancel 语义，见模块 docstring）------------------- def cancel(self, task_id): """把 queued/running 任务标 interrupted；已终态或不存在返回 False。 正在跑的子进程要真杀：先杀进程组（见 _kill_proc_group）， 再标状态；已终态或不存在返回 False。 """ task = self.get_task(task_id) if task is None or task["status"] in TERMINAL_STATUSES: return False self._cancel_requested.add(task_id) entry = self._running_snapshot().get(task_id) or {} if entry.get("proc") is not None: self._kill_proc_group(entry["proc"]) try: entry["proc"].wait(timeout=3) except Exception: pass # Docker 取消：按任务派生的容器名真杀，不只改数据库状态。 # 同步路径（runner 阻塞在 docker run 里）下 proc 为空， # 此时靠容器名杀容器；kill 尽力而为，失败也不影响后续 # 的状态标记。 if entry.get("container") is not None and \ hasattr(self.executor, "kill_container"): try: self.executor.kill_container(entry["container"]) except Exception: pass self._running_clear(task_id) conn = _db_connect(self.db_path) try: cur = conn.execute( "UPDATE tasks SET status = 'interrupted', updated_at = ?" " WHERE id = ? AND status NOT IN" " ('completed', 'failed', 'interrupted', 'merged'," " 'merge-failed')", (time.time(), task_id), ) conn.commit() claimed = cur.rowcount == 1 finally: conn.close() if not claimed: # 并发竞态：run_one 的轮询已看到本次取消的标记、先一步 # 把任务标成 interrupted——中断确实由本次取消触发，返回 True。 current = self.get_task(task_id) if current is not None and current["status"] == "interrupted": claimed = True if claimed: self._audit(task["project_id"], "worker_cancelled", task_id) log_event(self.logger, "worker_cancelled", task_id=task_id, project_id=task["project_id"], status="interrupted") return claimed def cleanup_stale_containers(self): """启动清理兜底：清掉上一轮残留的 codex- 容器。 只在执行器有 cleanup_stale_containers 时做（Docker）； 列出或删除失败只记日志，绝不抛异常炸掉 worker。 返回已清理的容器名列表（非 Docker 执行器返回 []）。 """ cleanup = getattr(self.executor, "cleanup_stale_containers", None) if cleanup is None: return [] try: cleaned = cleanup() except Exception as exc: log_event(self.logger, "worker_cleanup_failed", message="cleanup stale containers failed: %s" % exc) return [] if cleaned: log_event(self.logger, "worker_cleanup", message="cleaned stale containers: %s" % ",".join(cleaned)) return cleaned def _was_cancelled(self, task_id): if task_id in self._cancel_requested: return True task = self.get_task(task_id) return task is not None and task["status"] == "interrupted" # -- 缺 key 门 ------------------------------------------------------ def _executor_name(self): """执行器名：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；值本身绝不返回。""" try: value = self._env.get(self.env_key) except Exception: return False return value is not None and value != "" def _use_popen(self): """是否走进程模式：注入了 process_factory，或执行器是 默认 runner 的 LocalExecutor（真实 subprocess）。""" if self.process_factory is not None: return True return (isinstance(self.executor, LocalExecutor) and getattr(self.executor, "runner", None) is _sandbox_default_runner and hasattr(self.executor, "popen_task")) def _execute_via_process(self, task, kind="dispatch", deadline_seconds=None): """进程模式执行一轮：启动、轮询、取消时杀进程组。 返回与同步路径同款 result dict（status/stdout/stderr/ host_status/returncode）。取消时返回 status='cancelled' 的标记，调用方据此保持 interrupted；deadline_seconds 到期时杀进程组并返回 status='timeout' 的标记，调用方 据此置 failed（超时是失败，不是取消）。 """ task_id = task["id"] if self.process_factory is not None: proc = self.process_factory(task) else: proc = self.executor.popen_task(task, kind=kind) self._running_set(task_id, proc=proc) try: self._set_child_pid(task_id, getattr(proc, "pid", None)) except Exception: pass cancelled = False timed_out = False started = time.time() try: while proc.poll() is None: if task_id in self._cancel_requested or self._loop_stop: cancelled = True self._kill_proc_group(proc) break if deadline_seconds is not None and \ time.time() - started >= deadline_seconds: timed_out = True self._kill_proc_group(proc) break time.sleep(0.05) try: out, err = proc.communicate(timeout=5) except Exception: try: proc.kill() except Exception: pass out, err = proc.communicate() finally: self._running_clear(task_id) if cancelled: return {"executor": "local", "task_id": task_id, "status": "cancelled", "stdout": out or "", "stderr": err or "", "returncode": proc.returncode, "host_status": None} if timed_out: return {"executor": "local", "task_id": task_id, "status": "timeout", "stdout": out or "", "stderr": err or "", "returncode": proc.returncode, "host_status": None} fake = SimpleNamespace(returncode=proc.returncode, stdout=out or "", stderr=err or "") if kind == "goal" and hasattr( self.executor, "_goal_result_from_runner"): return self.executor._goal_result_from_runner(task_id, fake) if hasattr(self.executor, "_result_from_runner"): return self.executor._result_from_runner(task_id, fake) return {"executor": "local", "task_id": task_id, "status": "completed" if proc.returncode == 0 else "failed", "stdout": out or "", "stderr": err or "", "returncode": proc.returncode, "host_status": None} # -- 多租户隔离 ---------------------------------------------------- def _tenant_safe_name(self, project_id): """项目 id 的安全目录名：与 repos.checkout_dir 同规则。""" if project_id is None: raise ValueError("project_id 缺失，无法确定租户目录") raw = str(project_id) safe = re.sub(r"[^A-Za-z0-9_.-]", "-", raw).strip(".") if not safe or safe in (".", ".."): raise ValueError("project_id 无法映射为安全的租户目录名: %r" % raw) return safe def _tenant_dirs(self, project_id): """某项目的租户宿主目录（不存在时创建并播种）。 返回 dict：work / codex_home / repo_checkout / test_repo （均为宿主绝对路径）。播种规则： - codex-home：只复制模板的 config.toml（relay 端点与 provider 名，不含密钥值）；模板缺 config.toml 时直接 报错——空 home 会让 codex 去 api.openai.com 且无 auth， 绝不能静默放行（见 PR #18 的教训）。 - test-repo：整目录复制模板（它是 sample.py 夹具，含 .git 历史也无妨，本来就是公开的起始代码）。 - work / repo_checkout：只建目录。 并发下两个同项目任务同时播种是安全的：mkdir 用 exist_ok，复制前判存在且复制内容相同（模板），即使 race 到重复复制也不会写坏租户自己的数据（租户数据 在复制完成后才产生）。 """ import shutil safe = self._tenant_safe_name(project_id) base = os.path.join(self.tenants_root, safe) dirs = { "work": os.path.join(base, "work"), "codex_home": os.path.join(base, "codex-home"), "repo_checkout": os.path.join(base, safe), "test_repo": os.path.join(base, "test-repo"), } # 注意绝不 mkdir repo_checkout：repo_url 任务的检出由 # resolve_repo 负责 clone，空目录先占位会导致它误判 # “目录已存在但不是 git 仓库”而拒绝 clone；_tenant_mounts # 里按该目录是否存在决定是否挂载它（不存在说明任务 # 不用它，走 test-repo）。 for key in ("work", "codex_home", "test_repo"): os.makedirs(dirs[key], exist_ok=True) home_config = os.path.join(dirs["codex_home"], "config.toml") if not os.path.exists(home_config): template = os.path.join(self.tenant_template_home, "config.toml") if not os.path.exists(template): raise RuntimeError( "租户 codex-home 模板缺 config.toml：%s（空 home" " 会让 codex 去 api.openai.com 且无 auth，拒绝" " 派发）" % template) shutil.copyfile(template, home_config) if not os.path.exists(os.path.join(dirs["test_repo"], ".git")) \ and not os.path.exists(dirs["test_repo"] + "/sample.py"): # 目标为空目录（刚 mkdir 出来的）才播种；已有内容 # 时绝不覆盖（那是该项目自己的改动或上次播种结果）。 if not os.listdir(dirs["test_repo"]): if not os.path.exists(self.tenant_template_repo): raise RuntimeError( "租户 test-repo 模板不存在：%s" % self.tenant_template_repo) os.rmdir(dirs["test_repo"]) shutil.copytree(self.tenant_template_repo, dirs["test_repo"]) return dirs def _tenant_mounts(self, project_id): """某项目任务的 Docker 挂载列表（字符串，供执行器解析）。 容器内路径与非租户模式完全一致，只是宿主源按项目分开： /app/work、codex-home、repos/、test-repo 全部指向 该项目的租户目录；代码目录 /app:ro 由执行器基础挂载 提供，不在这里重复。静态 --docker-extra-mounts 在租户 模式下被这份列表替换（否则共享挂载会继续泄漏）。 repos 挂载只在检出目录已存在时才加（repo_url 任务在 派发前已由 _apply_project_repo clone 好；test-repo 任务 不用它，不存在时硬挂载会被执行器的存在性检查拒绝）。 """ dirs = self._tenant_dirs(project_id) mounts = [ "%s:/app/work:rw" % dirs["work"], "%s:/var/lib/codex-app-server/codex-home:rw" % dirs["codex_home"], "%s:/var/lib/codex-app-server/test-repo:rw" % dirs["test_repo"], ] if os.path.exists(dirs["repo_checkout"]): mounts.append( "%s:/var/lib/codex-app-server/repos/%s:rw" % (dirs["repo_checkout"], self._tenant_safe_name(project_id))) return mounts def _tenant_network(self, project_id): """某项目的 Docker 网络名（容器跨项目网络隔离用）。""" return "codex-" + self._tenant_safe_name(project_id) # -- 项目仓库 ------------------------------------------------------ def _project_repo_url(self, project_id): """查任务所属项目的 repo_url；查不到项目、列缺失、表缺失时 一律返回 None（走原有默认 repo，不报错、不炸掉派发）。 repo_url 本身只在 projects 表里，绝不从任务 prompt 里猜。""" try: conn = _db_connect(self.db_path) try: row = conn.execute( "SELECT repo_url FROM projects WHERE id = ?", (project_id,)).fetchone() finally: conn.close() except Exception: return None if row is None: return None try: url = row["repo_url"] except (KeyError, IndexError, TypeError): return None return url or None def _apply_project_repo(self, task): """派发前按项目 repo_url 解析仓库，返回错误文本或 None。 repo_url 为空：非租户模式不动 task，返回 None（走原有 默认 repo）；租户模式则把 task["repo"] 指到该项目的 test-repo 副本（容器内路径与原来一致，宿主源已按项目 分开），否则容器会 fallback 到共享 test-repo。 repo_url 非空：调 cloud.repos.resolve_repo 得到本地检出， 设 task["repo"] 再交给执行器——LocalExecutor 与 DockerExecutor 的 dispatch_task 都已读 task["repo"]，同一条 路径，不另起分支。租户模式下检出放在租户目录，task ["repo"] 设为容器内路径（挂载目标），而不是宿主检出 路径（worker 侧不再用 task["repo"] 当宿主路径）。 resolve 失败返回错误文本，调用方据此把任务置 failed （仓库解析失败绝不标 completed，也不在此重试，避免对 不可用的仓库反复 clone 烧时间）。 """ use_tenant = self.tenant_isolation and \ isinstance(self.executor, DockerExecutor) repo_url = self._project_repo_url(task.get("project_id")) if not repo_url: if use_tenant: try: self._tenant_dirs(task.get("project_id")) except (ValueError, RuntimeError) as exc: return "租户目录准备失败: %s" % exc task["repo"] = "/var/lib/codex-app-server/test-repo" return None try: if use_tenant: safe = self._tenant_safe_name(task.get("project_id")) resolve_repo( repo_url, os.path.join(self.tenants_root, safe), task.get("project_id"), runner=self.repos_runner) task["repo"] = \ "/var/lib/codex-app-server/repos/%s" % safe else: task["repo"] = resolve_repo( repo_url, self.repos_dir, task.get("project_id"), runner=self.repos_runner) except (RepoError, ValueError) as exc: return "仓库解析失败: %s" % exc except Exception as exc: return "仓库解析失败: %s" % exc return None # -- 执行 ---------------------------------------------------------- def _run_attempt(self, task): """单次执行尝试（不含认领、key 门、重试、成本与最终审计）。 返回 (result, error)：result 是执行器结果 dict，error 是 runner 异常时的文本（无异常时 None）。Docker 路径下把 worker 级 task_timeout 与执行器级 docker_timeout 取较小 值作为本次 docker run 超时（docker_timeout 未设时取 task_timeout）；Local popen 路径下把 task_timeout 作为 轮询截止时间；Local 同步路径无法中途抢占，超时只能靠 执行器自身 timeout（见模块 docstring 的诚实边界）。 调用方负责：取消优先、超时转 failed、重试、成本与审计。 """ task_id = task["id"] result = None error = None # goal 模式：任务带 verify_command 时走 host.py goal 循环， # 不走普通 dispatch。goal 循环内部已多轮重试，worker 级 # 的已存在续跑（run）对它没有意义（goal 没有 run 语义）。 goal_mode = bool(task.get("verify_command")) # Docker 同步路径：dispatch 阻塞在 docker run 里时，取消 # 线程靠容器名杀容器（见 cancel()）。容器名按任务派生， # 与执行器本身无关，所以这里只记名不碰执行器状态。 if hasattr(self.executor, "container_name"): try: self._running_set( task_id, container=self.executor.container_name(task)) except Exception: pass # Docker 超时取较小值：两个超时都不应静默延长另一个。 # 注意绝不能改共享的 executor.timeout（并发下两个任务会 # 互相覆盖对方的超时）；超时只作为本次调用的参数传下去。 effective_docker_timeout = None is_docker = isinstance(self.executor, DockerExecutor) if is_docker and self.task_timeout is not None: if self.executor.timeout is None: effective_docker_timeout = self.task_timeout else: try: effective_docker_timeout = min( float(self.task_timeout), float(self.executor.timeout)) except (TypeError, ValueError): effective_docker_timeout = self.task_timeout try: if self._use_popen(): result = self._execute_via_process( task, kind="goal" if goal_mode else "dispatch", deadline_seconds=self.task_timeout) combined = (result.get("stdout", "") or "") + \ (result.get("stderr", "") or "") if not goal_mode and result["status"] not in ( "completed", "cancelled", "timeout") and \ _ALREADY_EXISTS_MARKER in combined: result = self._execute_via_process( task, kind="run", deadline_seconds=self.task_timeout) else: if is_docker: tenant_mounts = None tenant_network = None proxy_env = None if self.tenant_isolation: try: tenant_mounts = self._tenant_mounts( task.get("project_id")) tenant_network = self._tenant_network( task.get("project_id")) except (ValueError, RuntimeError) as exc: # 提前返回会跳过下面的 finally（清容器 # 标记），这里先手动清掉，否则 _running # 里永远残留该任务，心跳与退出清理 # 都会把它当在跑任务。 self._running_clear(task_id) return ({"executor": "docker", "task_id": task_id, "status": "failed", "stdout": "", "returncode": None, "host_status": None, "stderr": ""}, "租户隔离准备失败: %s" % exc) # 网络建失败绝不降级默认网络跑容器（降级会 # 让隔离悄悄失效）；直接让本次尝试 failed， # 调用方按普通失败处理（重试/fallback 照常）。 if not self.executor.ensure_network(tenant_network): self._running_clear(task_id) return ({"executor": "docker", "task_id": task_id, "status": "failed", "stdout": "", "returncode": None, "host_status": None, "stderr": ""}, "租户 Docker 网络创建失败: %s" % tenant_network) if self.egress_proxy_port: # 代理在宿主上，容器经所在网络的网关访问； # 租户模式用项目网络，非租户用默认 bridge。 gw = self.executor.network_gateway(tenant_network) if not gw: self._running_clear(task_id) return ({"executor": "docker", "task_id": task_id, "status": "failed", "stdout": "", "returncode": None, "host_status": None, "stderr": ""}, "egress 代理网关解析失败，拒绝无代理" "运行（避免绕过出口白名单）") proxy_url = "http://%s:%s" % ( gw, self.egress_proxy_port) proxy_env = { "HTTP_PROXY": proxy_url, "HTTPS_PROXY": proxy_url, "http_proxy": proxy_url, "https_proxy": proxy_url, "NO_PROXY": "localhost,127.0.0.1", "no_proxy": "localhost,127.0.0.1", } if goal_mode: result = self.executor.goal_task( task, timeout=effective_docker_timeout, extra_mounts=tenant_mounts, network=tenant_network, extra_env=proxy_env) else: result = self.executor.dispatch_task( task, timeout=effective_docker_timeout, extra_mounts=tenant_mounts, network=tenant_network, extra_env=proxy_env) else: if goal_mode: result = self.executor.goal_task(task) else: result = self.executor.dispatch_task(task) combined = (result.get("stdout", "") or "") + \ (result.get("stderr", "") or "") if not goal_mode and result["status"] != "completed" and \ _ALREADY_EXISTS_MARKER in combined: if is_docker: result = self.executor.run_task( task, timeout=effective_docker_timeout, extra_mounts=tenant_mounts if self.tenant_isolation else None, network=tenant_network if self.tenant_isolation else None, extra_env=proxy_env if self.egress_proxy_port else None) else: result = self.executor.run_task(task) except Exception as exc: # runner 异常也如实写回 failed result = {"executor": "local", "task_id": task_id, "status": "failed", "stdout": "", "returncode": None, "host_status": None, "stderr": ""} error = str(exc) finally: # 同步路径执行结束：清掉取消用的容器标记（取消路径里 # 若已清过，这里再清一次无害）。进程模式的 proc 由 # _execute_via_process 自己的 finally 清，这里只清 # 容器项：_running_clear 整个删掉该任务项会连 proc # 也一起丢，所以这里只在 proc 为空时清容器标记—— # 实际做法是：如果该任务项里没有 proc，就整项清除； # 有 proc（进程模式）时不动它，等它自己的 finally。 entry = self._running_snapshot().get(task_id) or {} if entry.get("proc") is None: self._running_clear(task_id) # Docker 超时标记：执行器已在超时时 kill+rm 容器并返回 # failed（stdout 含 timeout），这里把状态统一成 timeout， # 调用方据此置 failed 并记 worker_timeout 审计。 if result is not None and result.get("status") == "failed" and \ "docker run timeout" in (result.get("stdout", "") or ""): result = dict(result, status="timeout") return result, error def _attempt_error(self, result, error, token): """一次尝试失败的可读原因（供 fallback 两次原因拼接用）。 超时用统一的超时文本；runner 异常用异常文本；同步路径的 stdout 是 host.py 的 JSON 时取它的 text 字段（比整段 JSON 可读）；最后退到 host_status token 或通用文本。绝不返回 任何密钥（stdout 里本就不应含密钥，text 字段同样不含）。 """ if result is not None and result.get("status") == "timeout": return "task timeout after %s seconds" % self.task_timeout if error: return error stdout = (result or {}).get("stdout", "") or "" try: parsed = json.loads(stdout) except (TypeError, ValueError): parsed = None if isinstance(parsed, dict) and parsed.get("text"): return str(parsed["text"]) # dispatch 输出现在带 error 字段（host.py 从 thread_history # 库里捞回的失败原因，如 404）；text 为空时优先用它， # 否则两次失败的拼接里只剩 host_status=404 这种代号。 if isinstance(parsed, dict) and parsed.get("error"): err = parsed["error"] if isinstance(err, dict): err = err.get("message") or str(err) return str(err)[:1000] # host.py 崩溃时 stdout 常为空、真正的报错在 stderr（例如 # “缺少 --repo”、Read-only file system）；只用 stdout 时 # 最终错误会被写成通用的 host execution failed，排障无从 # 下手，所以 stdout 为空时取 stderr 尾部（实际报错行在尾）。 stderr = (result or {}).get("stderr", "") or "" if not stdout.strip() and stderr.strip(): return stderr.strip()[-1000:] if token is not None: return "host_status=%s" % token return stdout.strip() or "host execution failed" def _fallback_token(self, result, error): """一次失败尝试的可 fallback 信号。 host_status 的真实含义是 host.py 回合状态（completed / failed / timeout / interrupted），不是 HTTP 状态码—— 真实世界里 host.py 从不返回 status=\"404\"，只有单测里的 假 runner 才会这么写。所以这里不能把 host_status 原样 当 token：host_status 本身已是可 fallback 码（404/429/ 5xx/timeout，含历史假 runner 形态）时直接返回；否则 （最常见：host_status=\"failed\"）从回合 JSON 的 text/error 字段、stdout、stderr、runner 异常里找显式 HTTP 状态码 或“model not found”类文本（按 404 计），找到且可 fallback 才返回；找不到显式信号时返回 None，绝不凭 “host execution failed”这类通用文本猜测，否则任何执行 器故障都会被误当成模型故障换模型烧额度。thread/start 崩溃（host.py 非 0 退出、不打印 JSON）时 host_status 为 None，同样走这条提取路径。 """ host_status = (result or {}).get("host_status") if host_status is not None and is_fallbackable(host_status): return host_status combined_parts = [ (result or {}).get("stdout", "") or "", (result or {}).get("stderr", "") or "", error or "", ] combined = " ".join(combined_parts) if not combined: return None lowered = combined.lower() if "model not found" in lowered or "unknown model" in lowered \ or "no such model" in lowered: return "404" match = re.search( r"(?:http/|status[\"':\s]{0,10}|code[\"':\s]{0,10}|error[\"':\s]{0,10})" r"(404|429|500|502|503|504)\b", combined, re.IGNORECASE) if match: return match.group(1) match = re.search(r"\b(404|429|500|502|503|504)\b", combined) if match: return match.group(1) return None def run_one(self, task=None): """认领并执行一个任务，返回结果 dict（见模块 docstring）。 未派发时返回 {"dispatched": False, "reason": ...} 且不动任务； 派发后（即使缺 key 失败）返回 dispatched=True 与最终状态。 """ if self._budget_exceeded(): return {"dispatched": False, "reason": "budget_exceeded"} if task is None: task = self.pick_next() if task is None: return {"dispatched": False, "reason": "no_ready_task"} task_id = 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", "%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() # 缺 key 门：认领之后、调 host.py 之前；缺 key 直接 failed， # 不调 runner（假 CLI 单测可断言 runner 未被调用）。 if not self.key_available(): error = _missing_key_error(self.env_key) self._set_status(task_id, "failed") self._write_execution_result( task_id, error_text=error, actual_model=self.model, started_at=started_at, finished_at=time.time()) self._audit(task["project_id"], "worker_missing_key", error) log_event(self.logger, "worker_missing_key", task_id=task_id, project_id=task["project_id"], status="failed") self._notify_task_finished( task_id, task["project_id"], "failed", error=error) return {"dispatched": True, "task_id": task_id, "host_name": self.executor._host_name(task) if hasattr(self.executor, "_host_name") else task_id, "status": "failed", "executor": self._executor_name(), "stdout": "", "error": error, "events": [], "cost_record_id": None} # 项目仓库解析：认领 + 缺 key 门之后、真正执行之前。 # repo_url 为空时不动（默认 repo）；解析失败时任务直接 failed， # 错误文本写清是仓库解析失败，绝不标 completed，也不在此重试。 repo_error = self._apply_project_repo(task) if repo_error is not None: self._set_status(task_id, "failed") self._write_execution_result( task_id, error_text=repo_error, actual_model=self.model, started_at=started_at, finished_at=time.time()) self._audit(task["project_id"], "worker_repo_failed", repo_error) log_event(self.logger, "worker_repo_failed", task_id=task_id, project_id=task["project_id"], status="failed") self._notify_task_finished( task_id, task["project_id"], "failed", error=repo_error) return {"dispatched": True, "task_id": task_id, "host_name": self.executor._host_name(task) if hasattr(self.executor, "_host_name") else task_id, "status": "failed", "executor": self._executor_name(), "stdout": "", "error": repo_error, "events": [], "cost_record_id": None} events = [] fallback_count = 0 result = None error = None retry_count = 0 # 最后真正跑过的模型：主模型起手，fallback 后换成 fallback 模型。 actual_model = self.model fallback_model = None first_error = None # 模型经任务的 model 字段传给执行器（执行器再经 host.py 的 # 顶层 --model 参数传给 dispatch/run）；不设时 host.py 用 # 配置文件里的 model。这里显式设成 worker 的主模型，让 # --model 覆盖配置真正生效，fallback 时再换成新模型。 task = dict(task) task["model"] = self.model # 有限重试与真实 fallback 的优先级：取消不重试；一次尝试失败 # 且可 fallback、且显式配了 fallback_models 时，优先真换模型 # 再跑一次（最多 1 次，同一任务、同一 worktree/分支，dispatch # 报“任务已存在”时 _run_attempt 内部走现有的 run 续跑路径）， # 第二次的结果就是最终结果，不再做同模型重试；未配 # fallback_models 时保持老行为（router 给出下一模型时只记事件 # 不重试）；其它失败（含 worker 超时）按 max_retries 做同模型 # 重试，退避等待后重跑一次 _run_attempt。 for attempt in range(self.max_retries + 1): result, error = self._run_attempt(task) if self._was_cancelled(task_id) or \ result.get("status") == "cancelled": break if task.get("verify_command"): # goal 模式只跑一次：循环内部已多轮重试（turn→验测→ # 喂回），worker 级重试/fallback 不再叠加——同任务重跑 # 会撞 host 任务名已存在，第二次尝试必失败且污染错误文本。 break if result.get("status") == "completed": break host_status_try = self._fallback_token(result, error) if host_status_try is None and \ result.get("status") == "timeout": # 进程/容器超时路径的 host_status 为空，超时本身就是 # 可 fallback 的失败 token（models.FALLBACKABLE 含它）。 host_status_try = "timeout" if fallback_model is None and self.fallback_models \ and host_status_try is not None \ and is_fallbackable(host_status_try): fb = self.fallback_models[0] record = dict(task) record["events"] = events # record_fallback 本身会 append 到 events， # 这里不要再 append 一次，否则事件会重复。 record_fallback( record, self.model, fb, str(host_status_try)) fallback_count = 1 self._audit(task["project_id"], "worker_fallback", "%s -> %s" % (self.model, fb)) first_error = self._attempt_error( result, error, host_status_try) fallback_model = fb actual_model = fb task = dict(task) task["model"] = fb result, error = self._run_attempt(task) break if self.router is not None and host_status_try is not None \ and is_fallbackable(host_status_try) and \ self.router.next_after( self.model, host_status_try) is not None: break if attempt < self.max_retries: retry_count += 1 self._audit( task["project_id"], "worker_retry", "%s: attempt %d failed, retry %d/%d" % (task_id, attempt + 1, retry_count, self.max_retries)) if self.retry_backoff_seconds: time.sleep(self.retry_backoff_seconds) continue break # 超时转 failed：超时是失败，不是取消；最终状态必须是 # VALID_STATUSES 里的 failed，绝不能把 "timeout" 写进库。 if result is not None and result.get("status") == "timeout": result = dict(result, status="failed") error = "task timeout after %s seconds" % self.task_timeout self._audit(task["project_id"], "worker_timeout", task_id) # 取消优先：执行期间被 cancel 的任务保持 interrupted， # 绝不用执行结果覆盖它（见模块 docstring 的取消说明）。 if self._was_cancelled(task_id) or result.get("status") == "cancelled": self._set_status(task_id, "interrupted") self._write_execution_result( task_id, result_text=(result or {}).get("stdout", ""), error_text="cancelled", actual_model=actual_model, started_at=started_at, finished_at=time.time()) log_event(self.logger, "worker_cancelled", task_id=task_id, project_id=task["project_id"], status="interrupted") return {"dispatched": True, "task_id": task_id, "host_name": self.executor._host_name(task) if hasattr(self.executor, "_host_name") else task_id, "status": "interrupted", "executor": self._executor_name(), "stdout": (result or {}).get("stdout", ""), "error": "cancelled", "events": [], "cost_record_id": None} stdout = result.get("stdout", "") or "" host_status = result.get("host_status") final = result["status"] if final != "completed": if fallback_model is not None: # 真 fallback 后第二次也失败：错误文本必须同时写清 # 两次原因，不能只留第二次的（超时转换已在前面把 # error 置成超时文本，那时 second_error 就是它）。 second_error = error or self._attempt_error( result, None, host_status) error = ("primary model %s failed: %s; " "fallback model %s failed: %s" % (self.model, first_error, fallback_model, second_error)) else: if result.get("goal_error"): error = result["goal_error"] if error is None: stderr = (result or {}).get("stderr", "") or "" error = stdout.strip() or \ (stderr.strip()[-1000:] if stderr.strip() else "") \ or "host execution failed" if retry_count: error = "[retries=%d] %s" % (retry_count, error) # 未配 fallback_models 时的老行为：只记事件、不重试。 # 注意用提取出的 token（真实回合失败时 host_status # 只是 "failed"，真正的 404 藏在回合 JSON 里）， # 不能直接用原始 host_status 判断。 _token = self._fallback_token(result, error) if self.router is not None and _token is not None \ and is_fallbackable(_token): nxt = self.router.next_after(self.model, _token) if nxt is not None: record = dict(task) record["events"] = events # record_fallback 本身会 append 到 events， # 这里不要再 append 一次，否则事件会重复。 record_fallback( record, self.model, nxt, str(_token)) fallback_count = 1 self._audit(task["project_id"], "worker_fallback", "%s -> %s" % (self.model, nxt)) final_written = self._set_status_if_running(task_id, final) finished_at = time.time() self._write_execution_result( task_id, result_text=stdout, error_text=error, actual_model=actual_model, started_at=started_at, finished_at=finished_at) with self._db_lock: self.usage_turns += 1 # 成本：host 协议不暴露真实 token 数，用 cost.py 的估算结构 # 记 estimated，绝不标 actual；fallback_count 复用同一记录； # 模型记最后真正跑的那个（fallback 成功时就是 fallback 模型）。 # cost 记录共用一条连接，并发下必须持锁写，否则两个线程 # 共用同一 sqlite 连接会直接报错。 total = estimate_tokens(task.get("prompt") or "", stdout) cost_id = self.costs.record( task_id, task["project_id"], actual_model, input_tokens=total, output_tokens=0, usd_micros=None, source="estimated", fallback_count=fallback_count) if final_written: self._audit(task["project_id"], "worker_completed" if final == "completed" else "worker_failed", task_id) log_event(self.logger, "worker_completed" if final == "completed" else "worker_failed", task_id=task_id, project_id=task["project_id"], status=final) if result.get("goal_met") is False: # goal 未达只发 goal-unmet，不再重复发 failed： # 同一结局两条通知会刷屏，digest 偏好下两者又都是 # 即时事件，更不能双发。 self._notify_task_finished( task_id, task["project_id"], "goal-unmet", error=error) elif final in ("completed", "failed"): self._notify_task_finished( task_id, task["project_id"], final, error=error) else: # 已被取消/回收：DB 状态以 interrupted 为准，不再 # 补一条 worker_completed/failed 审计误导排查。 final = "interrupted" self._audit(task["project_id"], "worker_interrupted", task_id) return {"dispatched": True, "task_id": task_id, "host_name": self.executor._host_name(task) if hasattr(self.executor, "_host_name") else task_id, "status": final, "executor": self._executor_name(), "stdout": stdout, "error": error, "events": events, "actual_model": actual_model, "cost_record_id": cost_id} def run_all(self, max_tasks=None): """循环认领执行，直到无就绪任务、预算超限或达到 max_tasks。 每轮都重新读库，所以上一轮刚 completed 的任务会解锁它的 依赖任务；每轮至少推进一个任务到终态，不会空转死循环。 capacity<=1 时走旧串行循环（行为与原来完全一致）。 capacity>1 时用线程池并行跑：同一 worker 内最多 capacity 个任务同时执行（capacity 从此是真实并发上限， 不再只是注册表展示）。认领本身是 CAS，两个线程同时 盯上同一任务时只有一个能拿到它，另一个的 run_one 返回 already_claimed，不计入结果、不死循环。依赖解锁靠 “等一个在跑任务完成后再认领”：pending 非空但暂无就绪 任务时，等一个完成再继续，而不是直接退出。 停机信号（_loop_stop/stop_event）只阻止新认领，不会 丢弃已提交的任务；排空等待由 run_loop 的 finally 负责。 """ if self.capacity <= 1: results = [] while max_tasks is None or len(results) < max_tasks: result = self.run_one() if not result["dispatched"]: break results.append(result) return results # capacity>1：手动线程池（不用 concurrent.futures）。 # 原因：本仓库有个单测把 cloud/ 插进 sys.path 后 # `import queue` 会解析到 cloud/queue.py 而不是标准库， # concurrent.futures 内部依赖标准库 queue.SimpleQueue， # 在该单测之后导入它会直接 AttributeError；手动用 # threading.Condition + 共享计数实现线程池，避开这个坑。 # 池线程数按队列深度自适应：每轮开始先数就绪任务， # 线程数 = min(capacity, 就绪数)——队列只有 1 个任务时 # 只开 1 个线程，不为空转线程池买单；队列深时开满 # capacity。就绪数是快照，任务完成解锁依赖时，池线程 # 靠“无就绪但仍有在跑任务则等待重试”继续接新任务， # 不会因为初始快照小就漏派。 # 池线程数 = capacity；每个池线程循环认领-执行，认领 # 本身是 CAS，两个线程盯上同一任务时只有一个能拿到。 # 依赖解锁：某线程暂无就绪任务时，若仍有线程在跑就等 # 条件变量（跑完的线程会 notify，可能解锁了依赖）， # 全部线程都空闲且无就绪任务时才退出。 # max_tasks 约束：已派发结果数 + 正在跑的任务数达到 # max_tasks 就不再认领，避免线程池提前提交导致超跑。 # 停机信号：只阻止新认领，不丢弃已在跑的任务；run_all # 在停机时立刻返回（后台线程继续跑），run_loop 的 # finally 做带超时的排空，超时后才杀残留。 state_lock = threading.Condition() shared = {"dispatched": 0, "active": 0, "stopped": False} results = [] results_lock = threading.Lock() def _pool_thread(): while True: with state_lock: if shared["stopped"] or self._loop_stop: return if max_tasks is not None and \ shared["dispatched"] + shared["active"] >= max_tasks: return try: task = self.pick_next() except Exception as exc: # 认领读库失败（如缺表、锁冲突）不炸池线程： # 等一轮再试，与 run_loop 的容错一致。 log_event(self.logger, "worker_pool_pick_error", message="worker pool pick error: %s" % exc) state_lock.wait(timeout=0.2) continue if task is None: if shared["active"] > 0: state_lock.wait(timeout=0.2) continue return shared["active"] += 1 result = {"dispatched": False} try: result = self.run_one(task) except Exception as exc: # 单个任务的意外异常绝不能拖垮整个池（串行 # 模式下 run_loop 也是这么容错的）；该任务已 # 认领为 running，之后由自动陈旧回收处理。 log_event(self.logger, "worker_pool_task_error", message="worker pool task error: %s" % exc) finally: with state_lock: shared["active"] -= 1 if result["dispatched"]: shared["dispatched"] += 1 state_lock.notify_all() if result["dispatched"]: with results_lock: results.append(result) try: ready_count = len(self._ready_tasks()) except Exception: # 深度快照失败时退回开满线程，池线程自己的认领 # 失败容错会处理，不让一次读库失败炸掉整轮。 ready_count = self.capacity threads = [threading.Thread(target=_pool_thread, daemon=True) for _ in range(min(self.capacity, max(1, ready_count)))] for t in threads: t.start() for t in threads: # 正常情况等全部池线程结束；停机信号来时不等了， # 立刻返回给 run_loop 做排空（后台线程是 daemon， # 且 run_loop finally 会杀残留，不会悬着进程）。 while t.is_alive(): t.join(timeout=0.2) if self._loop_stop: break if self._loop_stop: break return results def run_loop(self, interval=300, max_seconds=None, stop_event=None, drain_timeout=60, sweep_threshold=300): """常驻循环：每 interval 秒跑一轮 run_all，空闲也不退出。 退出条件：stop_event 被置位、_loop_stop 被信号处理置位、 或 max_seconds 到期。退出时先优雅排空：等正在跑的任务 自然结束（最多 drain_timeout 秒），超时后才杀残留进程组/ 容器并标 interrupted——以前是退出时立刻杀，部署重启会 无谓中断一个 16 秒就能跑完的任务。 运行中保障（与并发一起上的两件套）： - 心跳线程：每 30 秒给所有正在跑的任务刷 last_heartbeat， 否则长任务（>300 秒）在跑也会被陈旧回收误杀。 - 自动陈旧回收：每轮结束调一次 sweep_stale_tasks（阈值 sweep_threshold，默认 300 秒），把死 worker 遗留的 running 任务标 interrupted 并审计；活任务因心跳新鲜 绝不会被扫到。回收失败只记日志，不炸循环。 """ # 常驻启动清理兜底：清掉上一轮残留的 codex- 容器， # 失败只记日志（见 cleanup_stale_containers），不炸循环。 self.cleanup_stale_containers() # 运维心跳：启动时先写一次，之后每轮结束再写，server 的 # /health 据此算 worker_ok（见 server.py 的健康检查）。 self._write_heartbeat() # 多 worker 池注册：启动时注册一次，之后每轮结束写注册表 # 心跳（见本轮结束处）；注册失败只记日志，不炸循环。 self.register() started = time.time() all_results = [] # 心跳线程：daemon，run_loop 退出即停，不阻塞进程退出。 hb_stop = threading.Event() def _hb_loop(): while not hb_stop.is_set() and not self._loop_stop: if stop_event is not None and stop_event.is_set(): break self._heartbeat_running_tasks() hb_stop.wait(30) hb_thread = threading.Thread(target=_hb_loop, daemon=True) hb_thread.start() try: while not self._loop_stop: if stop_event is not None and stop_event.is_set(): break if max_seconds is not None and \ time.time() - started >= max_seconds: break try: results = self.run_all() except Exception as exc: # 常驻循环一轮炸了不带走整个 worker：记结构化日志， # 睡完本轮间隔再试下一轮（缺表、锁冲突都属此类）。 log_event(self.logger, "worker_loop_error", message="worker_loop_error: %s" % exc) results = [] all_results.extend(results) # 自动陈旧回收：死 worker 的 running 任务在这里被 # 标 interrupted；活任务心跳新鲜，不会被扫到。 if sweep_threshold is not None: try: from cloud.server import sweep_stale_tasks sweep_stale_tasks(self.db_path, sweep_threshold) except Exception as exc: log_event(self.logger, "worker_sweep_failed", message="worker sweep failed: %s" % exc) # 每轮结束写心跳：哪怕本轮一个任务都没派，worker 活着 # 这件事本身也要被 server 看到（文件心跳 + 注册表心跳）。 self._write_heartbeat() self._registry_heartbeat() # 按 0.2 秒切片睡，期间可响应 stop_event/_loop_stop。 deadline = time.time() + interval while time.time() < deadline: if self._loop_stop: break if stop_event is not None and stop_event.is_set(): break if max_seconds is not None and \ time.time() - started >= max_seconds: break time.sleep(0.2) finally: hb_stop.set() # 优雅排空：先等正在跑的任务自然结束，超时后才杀。 # 串行模式下 run_all 返回时任务已跑完，这里立刻通过； # 并行模式下停机信号退出时后台线程可能还在跑，先等 # drain_timeout 秒，超时后按下面杀残留。 if drain_timeout is not None: self._wait_for_running(timeout=drain_timeout) for task_id, entry in self._running_snapshot().items(): if entry.get("proc") is not None: self._kill_proc_group(entry["proc"]) try: entry["proc"].wait(timeout=3) except Exception: pass if entry.get("container") is not None and \ hasattr(self.executor, "kill_container"): try: self.executor.kill_container(entry["container"]) except Exception: pass self._running_clear(task_id) self._set_status(task_id, "interrupted") return all_results def _build_cli_executor(args, host_py): """CLI 的执行器构造：local 时返回 None（Worker 默认 LocalExecutor）， docker 时返回 DockerExecutor（env_key 名用 Worker 的解析规则， 值本身绝不从这里传——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( host_py, config_path=args.config, image=args.docker_image, cpus=args.docker_cpus, memory=args.docker_memory, user=args.docker_user, worktrees_root=args.docker_worktrees_root, env_key=resolve_env_key(args.env_key, args.codex_config), timeout=args.docker_timeout, extra_mounts=args.docker_extra_mounts, security_opts=args.docker_security_opt, pids_limit=args.docker_pids_limit, runtime=getattr(args, "docker_runtime", None)) def main(argv=None): """worker CLI：默认单次 run_all；--loop 进常驻循环。 只打印任务结果摘要（状态、host 名、错误文本），绝不打印 任何环境变量的值。 """ parser = argparse.ArgumentParser(description="cloud worker") parser.add_argument("--db", default="cloud-control-plane.db") parser.add_argument("--host-py", default=None, help="host.py 路径，缺省为 cloud/ 上一级的 host.py") parser.add_argument("--config", default=None) parser.add_argument("--repo", default=None) parser.add_argument("--heartbeat-file", default=None, help="运维心跳文件路径（默认：环境变量 " "CLOUD_WORKER_HEARTBEAT_FILE，未设则为 " "./worker-heartbeat.json；主机部署时用此" "参数指到 /var/lib/codex-app-server/ 下，" "与 server 的 health 读取路径保持一致）") parser.add_argument("--repos-dir", default="./repos", help="项目仓库检出根目录（默认 ./repos）；项目有" " repo_url 时 worker 在此 clone/fetch 各项目" "仓库；Docker 执行器下它必须放在" " --docker-extra-mounts 已挂载的目录下，" "否则容器内 task repo 路径不存在") parser.add_argument("--tenant-isolation", action="store_true", help="多租户隔离（仅 Docker 执行器）：每个项目用" "独立的 tenants_root/<项目>/work、codex-home、" "test-repo 副本与项目独立 Docker 网络，容器" "内路径不变，宿主挂载源按项目分开") parser.add_argument("--tenants-root", default="/var/lib/codex-app-server/tenants") parser.add_argument("--tenant-template-home", default="/var/lib/codex-app-server/codex-home", help="租户 codex-home 播种模板：只取其中的" "config.toml，绝不复制 thread_history") parser.add_argument("--tenant-template-repo", default="/var/lib/codex-app-server/test-repo", help="租户 test-repo 播种模板（整目录复制）") parser.add_argument("--env-key", default=None) parser.add_argument("--codex-config", default=None, help="含 env_key 行的 codex 配置文件（只读名字）") parser.add_argument("--model", default=None) parser.add_argument("--fallback-models", default=None, help="真实 fallback 换模型列表，逗号分隔" "（默认空 = 关闭，只记事件不换模型；" "配了时主模型可 fallback 失败后最多换" "列表里的第一个模型再跑一次，401/400" "绝不 fallback，取消绝不 fallback）") 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", "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 并发上限：同一 worker 内最多同时跑" "几个任务（默认 1，即旧串行行为；>1 时用" "线程池并行，认领仍是 CAS，成本写库持锁）") parser.add_argument("--docker-image", default="codex-app-server:latest") parser.add_argument("--docker-cpus", default="2") parser.add_argument("--docker-memory", default="2g") parser.add_argument("--docker-user", default="1000:1000") parser.add_argument("--docker-worktrees-root", default=None) parser.add_argument("--docker-timeout", type=float, default=None) parser.add_argument("--docker-extra-mounts", default=None, help="docker 额外挂载，逗号分隔 host:container[:ro|rw]，" "用于容器内 host.py 写 work/、host-state.json，" "以及挂载原仓库和 codex-home（同路径挂载，" "容器内配置无需改路径）") parser.add_argument("--docker-security-opt", action="append", default=None, help="docker --security-opt，可重复（如" " apparmor=docker-codex；no-new-privileges" " 已硬编码，重复传也不重复）") parser.add_argument("--docker-pids-limit", type=int, default=512, help="docker --pids-limit（默认 512，防 fork 炸弹；" "codex 工具链正常用量远低于此）") parser.add_argument("--docker-runtime", default=None, help="docker --runtime（不传 = docker 默认 runc；" "传 runsc = 任务容器走 gVisor 沙箱，需 daemon" "已注册该 runtime，否则起容器失败）") parser.add_argument("--egress-proxy-port", type=int, default=0, help="egress 过滤代理端口（0 = 不启用；生产配" " 8888）：启用后 Docker 任务容器的 HTTP(S)" " 出口全部走宿主该端口的代理，代理只放行" "白名单域名；网关查不到时任务直接 failed，" "绝不无代理运行") parser.add_argument("--claim-executor-only", action="store_true", help="严格 executor 认领：只认领 executor 恰好" "等于本 worker 类型的任务，绝不认领" " executor 为 NULL 的任务。宿主级沙箱" " worker 必须开，否则它会抢走本该进" " Docker 容器的 NULL 任务") parser.add_argument("--task-timeout", type=float, default=1800, help="每任务截止时间（秒，默认 1800）；超时后真杀" "容器/进程组、任务置 failed；与 --docker-timeout" "（单次 docker run 超时）取较小值，与" " --max-seconds（run_loop 整体墙钟上限）不同") parser.add_argument("--max-retries", type=int, default=1, help="模型任务失败重试次数（默认 1，0 表示不重试；" "取消与已记 fallback 的失败不重试）") parser.add_argument("--retry-backoff-seconds", type=float, default=5.0, help="重试前等待秒数（默认 5）") parser.add_argument("--loop", action="store_true", help="常驻循环，每 --interval 秒跑一轮") parser.add_argument("--interval", type=int, default=300) parser.add_argument("--drain-timeout", type=float, default=60, help="停机优雅排空等待上限（秒）：SIGTERM/SIGINT" "后先等在跑任务自然结束，超时后才杀残留" "并标 interrupted；默认 60 秒（systemd" "默认 90 秒停机时限内能完成）") parser.add_argument("--sweep-threshold", type=float, default=300, help="自动陈旧回收阈值（秒）：running 任务心跳" "超过该秒数未更新即被回收；活任务每 30" "秒刷心跳，不会被误杀") parser.add_argument("--max-seconds", type=float, default=None) args = parser.parse_args(argv) setup_logging() host_py = args.host_py if host_py is None: here = os.path.dirname(os.path.abspath(__file__)) host_py = os.path.join(os.path.dirname(here), "host.py") worker = Worker(args.db, host_py, config_path=args.config, repo=args.repo, env_key=args.env_key, codex_config_path=args.codex_config, model=args.model, fallback_models=( [m.strip() for m in args.fallback_models.split(",")] if args.fallback_models else []), max_turns=args.max_turns, executor=_build_cli_executor(args, host_py), task_timeout=args.task_timeout, max_retries=args.max_retries, retry_backoff_seconds=args.retry_backoff_seconds, repos_dir=args.repos_dir, heartbeat_file=args.heartbeat_file, worker_id=args.worker_id, capacity=args.worker_capacity, tenant_isolation=args.tenant_isolation, tenants_root=args.tenants_root, tenant_template_home=args.tenant_template_home, tenant_template_repo=args.tenant_template_repo, egress_proxy_port=args.egress_proxy_port, claim_executor_only=args.claim_executor_only) # CLI 启动清理兜底（常驻与单次都做）：清掉上一轮残留容器。 worker.cleanup_stale_containers() # CLI 起 worker 前先建表/迁移：全新库直接起循环不应因缺表而炸。 try: from cloud.server import init_db as _init_db _init_db(args.db) except Exception as exc: print("worker db 初始化失败：%s" % exc, file=sys.stderr) raise SystemExit(1) if args.loop: def _on_signal(signum, frame): worker._loop_stop = True signal.signal(signal.SIGTERM, _on_signal) signal.signal(signal.SIGINT, _on_signal) try: if args.loop: results = worker.run_loop(interval=args.interval, max_seconds=args.max_seconds, drain_timeout=args.drain_timeout, sweep_threshold=args.sweep_threshold) else: results = worker.run_all(max_tasks=args.max_tasks) finally: worker.close() print(json.dumps( [{"task_id": r.get("task_id"), "status": r.get("status"), "reason": r.get("reason"), "error": r.get("error")} for r in results], indent=2, ensure_ascii=False)) if __name__ == "__main__": main()