host
9f05178f2b
DockerExecutor gains a runtime parameter (passed as docker --runtime in build_run_command, after --pids-limit); worker CLI gains --docker-runtime, wired through _build_cli_executor. Unset or empty means no --runtime flag, so existing tasks keep docker default (runc); an unregistered runtime fails container start (fail closed), it never silently falls back. Tests: runsc appears in the command, default/empty omit it, CLI wiring passes through. README documents the flag and corrects a stale multi-tenant note (host kernel does not ban userns; bwrap fails inside Docker containers on nested-pidns proc mount).
1981 lines
100 KiB
Python
1981 lines
100 KiB
Python
"""云端 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: <env_key 名>`，与之前
|
||
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 已存在: <name>，先换名或 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"),
|
||
}
|
||
|
||
|
||
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/<project_id>/{work,codex-home,test-repo,
|
||
# <project_id>}，容器内路径保持与原来完全一致（/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 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"}.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 _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 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/<pid>、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
|
||
# 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, deadline_seconds=self.task_timeout)
|
||
combined = (result.get("stdout", "") or "") + \
|
||
(result.get("stderr", "") or "")
|
||
if 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",
|
||
}
|
||
result = self.executor.dispatch_task(
|
||
task, timeout=effective_docker_timeout,
|
||
extra_mounts=tenant_mounts,
|
||
network=tenant_network,
|
||
extra_env=proxy_env)
|
||
else:
|
||
result = self.executor.dispatch_task(task)
|
||
combined = (result.get("stdout", "") or "") + \
|
||
(result.get("stderr", "") or "")
|
||
if 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")
|
||
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")
|
||
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 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 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)
|
||
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()
|