"""沙箱执行器：可插拔的任务执行接口。 零第三方依赖，只用 Python 标准库；不重写 agent 逻辑。 - Executor：基类，子类必须实现 execute(task)。 - LocalExecutor：先行实现，复用 host.py 现有 worktree + codex 沙箱， 通过 host.py CLI 子进程做只读查询封装；不直接跑模型。 - DockerExecutor：已接通。docker run --rm，任务 worktree 挂载到 /work（可写）、代码目录挂载到 /app:ro（只读），--cpus/--memory 资源限额，--user 非 root，-e 只传 env_key 名（不带值，key 缺失 时不启动容器）；超时与取消时 docker kill + rm -f，不留僵尸 容器。网络白名单不在本类里假装实现（见 cloud/README.md）。 - CloudExecutor：占位，未接通云端执行。 """ import json import os import re import shutil import subprocess import sys MAX_STDOUT_CHARS = 2000 class Executor: """执行器基类。 task 是 dict，含 id、project_id、repo、prompt、branch（可选）。 execute(task) -> dict，至少含 executor、task_id、status、stdout。 """ def execute(self, task) -> dict: raise NotImplementedError("子类必须实现 execute(task)") def _truncate_stdout(text): if text is None: return "" return str(text)[:MAX_STDOUT_CHARS] def _default_runner(cmd): """subprocess.run 的薄封装：捕获输出、文本模式。""" return subprocess.run(cmd, capture_output=True, text=True) class LocalExecutor(Executor): """本地执行器：调 host.py CLI 子进程做执行封装。 真正跑模型回合的 dispatch/run 由控制面在确认模型 key 供给后才 调用，本模块只做执行器骨架和 CLI 调用封装。因此 execute() 在此 只发 host.py 的只读命令（例如 status），绝不在 execute 里直接 跑模型回合；需要跑模型时，控制面另行调用 host.py dispatch/run。 构造函数：LocalExecutor(host_py_path, config_path=None, runner=None)。 runner 默认是 subprocess.run 的薄封装；单测可注入假 runner，假 runner 返回 CompletedProcess 风格的 result 对象（含 returncode、 stdout），本类只读这两个字段，不依赖其它字段。 execute() 之外另有 dispatch_task()/run_task()：它们是 worker 派发时调 host.py 真实子命令的入口——dispatch 对应 host.py 的 dispatch（建任务并跑一轮 turn），run 对应 host.py 的 run （在已有 host 任务上再跑一轮 turn）。两者都只发真实存在的 host.py 子命令，不编造命令名；execute() 本身仍只发只读 status， 绝不在 execute 里直接跑模型回合（见 test_queue_sandbox 的断言）。 """ def __init__(self, host_py_path, config_path=None, runner=None, repo=None, model=None): self.host_py_path = host_py_path self.config_path = config_path self.runner = runner if runner is not None else _default_runner self.repo = repo # 模型：host.py 的 --model 是顶层全局参数（dispatch/run 共用）， # 不传时 host.py 用配置文件里的 model；传了就覆盖配置。 # worker 的 fallback 靠它把换模型后的第二次尝试真正发到新模型上， # 不改 host.py 的 dispatch/run 签名。 self.model = model def _base_command(self, model=None): cmd = [sys.executable, self.host_py_path] if self.config_path is not None: cmd += ["--config", self.config_path] effective = model or self.model if effective: cmd += ["--model", effective] return cmd def _command(self, task_id): cmd = self._base_command() cmd += ["status", "--task", str(task_id)] return cmd def _result_from_runner(self, task_id, result): """把 runner 返回值统一成 execute() 同款结果 dict。 host.py dispatch/run 成功时打印 JSON（含 status/text）， 失败时以 SystemExit(1) 退出并打印错误文本。因此：returncode 非 0 一律 failed；returncode 为 0 时优先取 JSON 里的 status （completed 外的状态，如 timeout，均按失败计，worker 侧再 判断是否可 fallback），取不到 JSON 时按 completed 计。 """ returncode = getattr(result, "returncode", 1) stdout = _truncate_stdout(getattr(result, "stdout", "")) status = "failed" if returncode == 0: status = "completed" try: parsed = json.loads(stdout) except (TypeError, ValueError): parsed = None if isinstance(parsed, dict) and parsed.get("status"): status = ("completed" if parsed["status"] == "completed" else "failed") host_status = parsed["status"] else: host_status = None else: host_status = None return { "executor": "local", "task_id": task_id, "status": status, "stdout": stdout, "returncode": returncode, "host_status": host_status, # host.py 的 RuntimeError（如“任务已存在”）走 traceback 到 # stderr，worker 判断 dispatch 续跑时需要它；execute() # 的旧结果 dict 不含此字段，保持不变。 "stderr": _truncate_stdout(getattr(result, "stderr", "")), } def _host_name(self, task): """cloud 任务 id -> host 任务名：只保留安全字符。 server 生成的 id 本就是 uuid hex，走这一步只是防御自定义 id； 同一 id 每次算出同一名字，dispatch 失败后的 run 续跑才能 找到同一个 host 任务。 """ raw = str(task.get("id")) return re.sub(r"[^A-Za-z0-9_.-]", "-", raw) def dispatch_task(self, task) -> dict: """调 host.py dispatch：建 host 任务并跑一轮 turn。 命令：host.py [--config CFG] [--model M] dispatch --name --task [--repo ]。repo 取任务的 repo 字段， 没有则取构造时传的 repo，都没有则不传（此时 host.py 用其 配置文件里的 repo，配置里也没有时 host.py 自己报错， worker 把该报错原样写回 failed，不编造成功）。模型取任务的 model 字段，没有则取构造时传的 model，都没有则不传 --model， host.py 用配置文件里的 model（fallback 时 worker 会把任务的 model 换成 fallback 模型，第二次 dispatch/run 就跑在新模型上）。 """ if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") if not task.get("prompt"): raise ValueError("任务缺少 prompt") cmd = self._base_command(model=task.get("model")) cmd += ["dispatch", "--name", self._host_name(task), "--task", task["prompt"]] repo = task.get("repo") or self.repo if repo: cmd += ["--repo", repo] return self._result_from_runner( task["id"], self.runner(cmd)) def run_task(self, task) -> dict: """调 host.py run：在已有 host 任务上再跑一轮 turn。 命令：host.py [--config CFG] [--model M] run --task --prompt 。worker 只在 dispatch 报“任务已存在”时 用它续跑，不用它做首次派发。模型与 dispatch_task 同规则： 任务的 model 字段优先，fallback 续跑时就是新模型。 """ if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") if not task.get("prompt"): raise ValueError("任务缺少 prompt") cmd = self._base_command(model=task.get("model")) cmd += ["run", "--task", self._host_name(task), "--prompt", task["prompt"]] return self._result_from_runner( task["id"], self.runner(cmd)) def popen_task(self, task, kind="dispatch"): """进程组启动 host.py 子进程，返回 Popen（供真取消用）。 start_new_session=True 让子进程自成进程组，worker 取消时 killpg 才能连同子孙一起杀掉。注意：只有默认 runner（真实 subprocess）时才用本方法；单测注入的假 runner 走同步路径， 不走这里。 """ if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") if not task.get("prompt"): raise ValueError("任务缺少 prompt") cmd = self._base_command(model=task.get("model")) if kind == "run": cmd += ["run", "--task", self._host_name(task), "--prompt", task["prompt"]] else: cmd += ["dispatch", "--name", self._host_name(task), "--task", task["prompt"]] repo = task.get("repo") or self.repo if repo: cmd += ["--repo", repo] return subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True) def execute(self, task) -> dict: task_id = task.get("id") if isinstance(task, dict) else None if task_id is None: raise ValueError("任务缺少 id") result = self.runner(self._command(task_id)) returncode = getattr(result, "returncode", 1) stdout = _truncate_stdout(getattr(result, "stdout", "")) return { "executor": "local", "task_id": task_id, "status": "completed" if returncode == 0 else "failed", "stdout": stdout, "returncode": returncode, } class DockerExecutor(Executor): """容器执行器（已接通，命令构造可测，真实 docker 需外部主机）。 设计（与 LocalExecutor 对齐，区别只在“进程跑在哪”）： - 容器内跑的仍是 host.py：代码目录以只读方式挂载到 /app， 容器内命令就是 python3 /app/host.py dispatch/run/status， 不重写 agent 逻辑；任务自己的 worktree 挂载到 /work（可写）， 绝不挂载宿主根目录、绝不挂载任何含密钥的目录。 - 容器名：codex-，host 任务名只保留安全字符， 同一任务每次算出同一名字，取消时才能按名杀容器。 - key：只把 env_key 名以 `-e <名>` 传给 docker（docker 从宿主 进程环境里取值注入容器），命令列表里绝不出现 key 的值； key 缺失时在调 runner 之前就明确失败，容器根本不启动。 本类只查 env 映射里的存在性与非空，绝不打印、记录 key 的值。 - 资源：--cpus / --memory 可配置，默认 2 核 / 2G，写进文档。 - 清理：docker run --rm，失败也 --rm；超时或取消时先 docker kill 再 docker rm -f，绝不留僵尸容器。 - 网络诚实边界：docker run 默认 bridge 网络；本类不做 egress 白名单——真正的“只放行模型 API”需要宿主防火墙/代理规则， 那是部署项，不是本类假装做到的（见 cloud/README.md）。 构造函数：DockerExecutor(host_py_path=None, config_path=None, runner=None, image="codex-app-server:latest", cpus="2", memory="2g", user="1000:1000", worktrees_root=None, env=None, env_key="XCM_API_KEY", timeout=None)。 host_py_path 缺省时取 cloud/ 上一级的 host.py（与 worker CLI 的缺省规则一致），以便 DockerExecutor() 无参调用仍可用。 runner 默认是带 timeout 的 subprocess 封装；单测注入的假 runner 可以只接受 cmd 一个参数（本类遇 TypeError 时回退 到只传 cmd 调用），也可以接受 (cmd, timeout=...)。 docker 二进制缺失时 raise RuntimeError("docker not available")， 绝不静默降级为本地执行。 """ def __init__(self, host_py_path=None, config_path=None, runner=None, image="codex-app-server:latest", cpus="2", memory="2g", user="1000:1000", worktrees_root=None, env=None, env_key="XCM_API_KEY", timeout=None, extra_mounts=None, model=None, security_opts=None, pids_limit=None, extra_env=None): if host_py_path is None: here = os.path.dirname(os.path.abspath(__file__)) host_py_path = os.path.join(os.path.dirname(here), "host.py") self.host_py_path = host_py_path self.config_path = config_path # 模型与 LocalExecutor 同规则：任务的 model 字段优先，没有则 # 取构造时传的 model，都没有则不传 --model（容器内 host.py # 用配置文件里的 model）。fallback 时 worker 换任务的 model。 self.model = model self.runner = runner if runner is not None else _docker_default_runner self.image = image self.cpus = str(cpus) self.memory = str(memory) self.user = user self.worktrees_root = worktrees_root self._env = env if env is not None else os.environ self.env_key = env_key self.timeout = timeout self.extra_mounts = self._parse_extra_mounts(extra_mounts) # 外层加固：security_opts 是 --security-opt 的值列表 # （如 apparmor=docker-codex；no-new-privileges 已硬编码， # 这里传了也不重复）；pids_limit 防 fork 炸弹（codex # 工具链正常用量远低于 512，见 worker CLI 默认值）； # extra_env 是 -e NAME=VALUE 形式的非密钥环境变量 # （如 egress 代理地址——绝不放密钥值，密钥只走 env_key # 的 -e 名方式）。 self.security_opts = list(security_opts or []) self.pids_limit = pids_limit self.extra_env = dict(extra_env or {}) @staticmethod def _parse_extra_mounts(extra_mounts): """解析额外挂载，格式 host:container 或 host:container:ro/rw。 安全红线（违反时直接抛错，绝不静默放行）：宿主路径必须 是绝对路径且必须已存在（文件挂载时 docker 对不存在路径 会自动建目录，极易把文件挂载变成目录挂载）；容器路径 必须是绝对路径，不能是 / 或 /app 本身（那会整体覆盖 只读代码挂载）；宿主路径不能是 /、/etc、/root，也不能 在 /etc/codex-app-server 之下（那里放着 key 文件）。 """ if not extra_mounts: return [] if isinstance(extra_mounts, str): extra_mounts = [m for m in extra_mounts.split(",") if m.strip()] out = [] for spec in extra_mounts: parts = spec.split(":") if len(parts) not in (2, 3): raise ValueError("extra mount 格式必须是 host:container[:ro|rw]: %s" % spec) host_p, cont_p = parts[0], parts[1] mode = parts[2] if len(parts) == 3 else "rw" if mode not in ("ro", "rw"): raise ValueError("extra mount 模式只能是 ro 或 rw: %s" % spec) if not os.path.isabs(host_p) or not os.path.isabs(cont_p): raise ValueError("extra mount 路径必须是绝对路径: %s" % spec) if host_p in ("/", "/etc", "/root") or host_p.startswith("/etc/codex-app-server"): raise ValueError("extra mount 拒绝危险宿主路径: %s" % spec) if cont_p in ("/", "/app"): raise ValueError("extra mount 拒绝覆盖容器根或 /app 本身: %s" % spec) if not os.path.exists(host_p): raise ValueError("extra mount 宿主路径不存在，拒绝挂载: %s" % spec) out.append((host_p, cont_p, mode)) return out # -- 名字 / 路径 --------------------------------------------------- def _host_name(self, task): raw = str(task.get("id")) return re.sub(r"[^A-Za-z0-9_.-]", "-", raw) def container_name(self, task): """任务派生的容器名：codex-<安全字符任务名>。""" return "codex-" + self._host_name(task) def _app_dir(self): """代码目录：host_py_path 所在的目录，挂载到 /app:ro。""" return os.path.dirname(os.path.abspath(self.host_py_path)) def _worktree_for(self, task): """任务自己的 worktree：显式字段优先，否则按派生目录。 显式字段 worktree > cwd；都没有时用 worktrees_root/ （未传 worktrees_root 时默认在 host.py 同级的 work/ docker-worktrees 下）。本方法不建目录：docker -v 对不存 在的宿主路径会自动建目录，测试里传显式 worktree 即可 避免任何副作用。 """ if task.get("worktree"): return task["worktree"] if task.get("cwd"): return task["cwd"] root = self.worktrees_root if root is None: root = os.path.join(self._app_dir(), "work", "docker-worktrees") return os.path.join(root, self._host_name(task)) # -- 前置检查 ------------------------------------------------------ def _check_docker(self): if shutil.which("docker") is None: raise RuntimeError("docker not available") def _check_key(self): """key 门：只查存在性与非空，值本身绝不返回、不打印。""" if not self.env_key: return try: value = self._env.get(self.env_key) except Exception: value = None if value is None or value == "": raise RuntimeError( "Missing environment variable: %s" % self.env_key) # -- 命令构造 ------------------------------------------------------ def build_run_command(self, task, inner_cmd, extra_mounts=None, network=None, extra_env=None): """构造完整的 docker run 命令（纯函数，可单测断言）。 顺序固定：docker run --rm --name <容器名> --user <用户> --cpus <核数> --memory <内存> [--cap-drop=ALL --security-opt=no-new-privileges] [--security-opt <自定义>] [--pids-limit ] [--network <网络>] -v :/work -v <代码目录>:/app:ro -w /work [-e ] [-e ...] <镜像> <容器内命令...>。 -e 只传名字，不带 =值（env_key）；extra_env 的 -e 带值， 只用于非密钥变量（如 egress 代理地址）。 extra_mounts 给定时替换构造时的静态挂载（多租户隔离 用：worker 按项目算出挂载后每次调用传入）；不给时用 静态挂载。network 给定时容器进该 Docker 网络（多租户 用项目独立网络，容器之间默认 bridge 上互相可见的问题 由此解决）；不给时用 docker 默认网络。 降权：容器默认已是非 root 用户，再加 --cap-drop=ALL 与 no-new-privileges（提权类工具调用会被挡下，codex 的常规文件/git 操作不受影响，见多租户实测）。 """ self._check_docker() self._check_key() cmd = ["docker", "run", "--rm", "--name", self.container_name(task), "--user", self.user, "--cpus", self.cpus, "--memory", self.memory, "--cap-drop=ALL", "--security-opt=no-new-privileges"] for opt in self.security_opts: if opt != "no-new-privileges": cmd += ["--security-opt", opt] if self.pids_limit is not None: cmd += ["--pids-limit", str(self.pids_limit)] if network: cmd += ["--network", network] cmd += ["-v", "%s:/work" % self._worktree_for(task), "-v", "%s:/app:ro" % self._app_dir(), "-w", "/work"] # 额外挂载放在基础挂载之后：/app/work 这类子路径挂载会 # 覆盖 /app:ro 的对应子路径，让容器内 host.py 能写 # work/ 和 host-state.json，同时代码主体仍只读。 mounts = self.extra_mounts if extra_mounts is None \ else self._parse_extra_mounts(extra_mounts) for host_p, cont_p, mode in mounts: cmd += ["-v", "%s:%s:%s" % (host_p, cont_p, mode)] if self.env_key: cmd += ["-e", self.env_key] merged_env = dict(self.extra_env) if extra_env: merged_env.update(extra_env) for name, value in sorted(merged_env.items()): cmd += ["-e", "%s=%s" % (name, value)] cmd += [self.image] cmd += list(inner_cmd) return cmd def _inner_host_command(self, *args, model=None): """容器内 host.py 命令：python3 /app/host.py [args...]。 config_path 若已给，容器内用 /app/ 指代它—— 调用方须保证配置文件就在代码目录里（host.py 配置约定 本就放在项目目录）；不在代码目录时不静默猜路径， 由 _inner_host_command 的调用方在传 config 前自行确认。 这里只做最简单的同名映射，不展开宿主绝对路径进容器。 --model 是 host.py 的顶层全局参数，必须放在子命令之前， 所以 model 在这里插到 --config 之后、子命令之前。 """ cmd = ["python3", "/app/host.py"] if self.config_path is not None: cmd += ["--config", "/app/" + os.path.basename(self.config_path)] effective = model or self.model if effective: cmd += ["--model", effective] cmd += list(args) return cmd # -- runner 调用 / 清理 -------------------------------------------- def _call_runner(self, cmd, timeout=None): """调 runner：优先传 timeout，假 runner 只接受 cmd 时回退。 timeout 参数只影响本次调用（并发下两个任务各用各的 超时）；不传时用实例的 self.timeout。绝不能在调用方 临时改 self.timeout 再恢复，那在并发下会互相覆盖。 """ try: return self.runner( cmd, timeout=self.timeout if timeout is None else timeout) except TypeError: return self.runner(cmd) def list_containers(self, prefix="codex-"): """列出名字以 prefix 开头的容器（默认 codex-），失败返回 []。 用 `docker ps -a --format {{.Names}}`，输出按行取容器名； runner 抛异常或输出不可解析时返回空列表，调用方（worker 启动清理）据此只记日志、不炸掉 worker。 """ try: result = self._call_runner( ["docker", "ps", "-a", "--format", "{{.Names}}"]) except Exception: return [] out = getattr(result, "stdout", "") or "" names = [] for line in str(out).splitlines(): name = line.strip() if name and name.startswith(prefix): names.append(name) return names # 任务容器名是 codex-<32 位 hex 任务 id>（见 _container_name）。 # 启动清理只能删这种容器，绝不能按 "codex-" 前缀全删：基础设施 # 容器也叫 codex-*（例如 codex-postgres），按前缀删会把数据库 # 容器当成残留任务容器杀掉（2026-10-26 实测：worker 重启时 # codex-postgres 就是这样被删掉，导致控制面连不上 Postgres）。 _TASK_CONTAINER_RE = re.compile(r"^codex-[0-9a-f]{32}$") def cleanup_stale_containers(self, prefix="codex-"): """清理上一轮残留的任务容器：只删 codex-<32 位 hex 任务 id>。 返回已清理的容器名列表；单个容器清理失败只跳过它（kill 本身尽力而为），列出失败时返回 []，绝不抛异常。 基础设施容器（如 codex-postgres）名字不匹配任务 id 正则， 永远不会被这里删掉。 """ cleaned = [] for name in self.list_containers(prefix=prefix): if not self._TASK_CONTAINER_RE.match(name): continue try: if self.kill_container(name): cleaned.append(name) except Exception: continue return cleaned def kill_container(self, task_or_name): """杀容器：docker kill + docker rm -f，尽力而为不抛异常。 task_or_name 可以是任务 dict（取容器名）或容器名字符串。 返回 True 表示至少 kill 或 rm 其中一步成功执行（runner 未抛异常），False 表示两步都失败或 docker 不存在。 取消路径调用它，绝不只改数据库状态。 """ name = (self.container_name(task_or_name) if isinstance(task_or_name, dict) else str(task_or_name)) ok = False for cmd in (["docker", "kill", name], ["docker", "rm", "-f", name]): try: self._call_runner(cmd) ok = True except Exception: continue return ok def _run_in_container(self, task, inner_cmd, timeout=None, extra_mounts=None, network=None, extra_env=None): """在容器里跑一轮内层命令，返回 LocalExecutor 同款结果 dict。 超时：runner 抛 subprocess.TimeoutExpired（或同名异常） 时先 kill+rm 容器，再返回 failed（stdout 截断 + timeout 说明，returncode 为 None）；其它 runner 异常也如实返回 failed，不编造成功。timeout 只影响本次调用，不改实例。 extra_mounts/network/extra_env 只影响本次调用（多租户 按项目传入、egress 代理按任务网络传入）。 """ task_id = task.get("id") if isinstance(task, dict) else None if task_id is None: raise ValueError("任务缺少 id") cmd = self.build_run_command(task, inner_cmd, extra_mounts=extra_mounts, network=network, extra_env=extra_env) try: result = self._call_runner(cmd, timeout=timeout) except Exception as exc: if exc.__class__.__name__ == "TimeoutExpired": self.kill_container(task) return { "executor": "docker", "task_id": task_id, "status": "failed", "stdout": "docker run timeout", "returncode": None, "host_status": None, "stderr": "", "container": self.container_name(task), } return { "executor": "docker", "task_id": task_id, "status": "failed", "stdout": _truncate_stdout(str(exc)), "returncode": None, "host_status": None, "stderr": "", "container": self.container_name(task), } fake = SimpleNamespaceShim(result) parsed = _parse_host_result(task_id, fake) parsed["executor"] = "docker" parsed["container"] = self.container_name(task) return parsed def dispatch_task(self, task, timeout=None, extra_mounts=None, network=None, extra_env=None) -> dict: """容器内调 host.py dispatch：建 host 任务并跑一轮 turn。""" if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") if not task.get("prompt"): raise ValueError("任务缺少 prompt") inner = self._inner_host_command( "dispatch", "--name", self._host_name(task), "--task", task["prompt"], model=task.get("model")) repo = task.get("repo") if repo: inner += ["--repo", repo] return self._run_in_container(task, inner, timeout=timeout, extra_mounts=extra_mounts, network=network, extra_env=extra_env) def run_task(self, task, timeout=None, extra_mounts=None, network=None, extra_env=None) -> dict: """容器内调 host.py run：在已有 host 任务上再跑一轮 turn。""" if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") if not task.get("prompt"): raise ValueError("任务缺少 prompt") inner = self._inner_host_command( "run", "--task", self._host_name(task), "--prompt", task["prompt"], model=task.get("model")) return self._run_in_container(task, inner, timeout=timeout, extra_mounts=extra_mounts, network=network, extra_env=extra_env) def network_gateway(self, network_name): """某 Docker 网络的网关 IP（宿主在该网桥上的地址）。 egress 代理跑在宿主上时，容器只能通过网关 IP 访问它 （容器内的 127.0.0.1 是自己，不是宿主）。返回 None 表示查不到（docker 不可用、网络不存在等），调用方 必须按失败处理，绝不能无代理跑容器（那样等于绕过 出口白名单）。网络名为空时查默认 bridge 网络。 """ name = network_name or "bridge" try: result = self._call_runner( ["docker", "network", "inspect", name, "--format", "{{(index .IPAM.Config 0).Gateway}}"], timeout=30) except Exception: return None try: if getattr(result, "returncode", 1) != 0: return None gw = (getattr(result, "stdout", "") or "").strip() except Exception: return None return gw or None def ensure_network(self, name): """确保 Docker 网络存在：不存在就创建，已存在就忽略。 返回 True 表示网络可用（创建成功或本来就存在）， False 表示创建失败（调用方继续派发，docker run 时会 因网络不存在而明确失败，不在这里静默降级默认网络—— 降级会让隔离悄悄失效，比失败更糟，所以这里只返回 布尔值，是否继续由调用方按失败处理：worker 在租户 模式下网络创建失败时直接让任务 failed，不跑容器）。 """ if not name: return False try: result = self._call_runner(["docker", "network", "create", name], timeout=30) except Exception as exc: return "already exists" in str(exc) try: if getattr(result, "returncode", 1) == 0: return True text = (getattr(result, "stderr", "") or "") + \ (getattr(result, "stdout", "") or "") return "already exists" in text except Exception: return False def execute(self, task) -> dict: """只读执行：容器内调 host.py status，不直接跑模型回合。 与 LocalExecutor.execute 对齐：execute 本身只发 status， 跑模型请走 dispatch_task/run_task（worker 派发路径）。 """ if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") inner = self._inner_host_command( "status", "--task", self._host_name(task)) return self._run_in_container(task, inner) class SimpleNamespaceShim: """把 runner 返回值统一成 returncode/stdout/stderr 三字段。""" def __init__(self, result): self.returncode = getattr(result, "returncode", 1) self.stdout = getattr(result, "stdout", "") self.stderr = getattr(result, "stderr", "") def _parse_host_result(task_id, fake): """LocalExecutor._result_from_runner 的无类版本，供 Docker 用。 host.py 成功时打印 JSON（含 status/text），失败时非 0 退出； returncode 非 0 一律 failed；为 0 时优先取 JSON 里的 status， completed 外的状态均按 failed 计。 """ returncode = getattr(fake, "returncode", 1) stdout = _truncate_stdout(getattr(fake, "stdout", "")) status = "failed" host_status = None if returncode == 0: status = "completed" try: parsed = json.loads(stdout) except (TypeError, ValueError): parsed = None if isinstance(parsed, dict) and parsed.get("status"): host_status = parsed["status"] status = ("completed" if host_status == "completed" else "failed") return { "executor": "docker", "task_id": task_id, "status": status, "stdout": stdout, "returncode": returncode, "host_status": host_status, "stderr": _truncate_stdout(getattr(fake, "stderr", "")), } def _docker_default_runner(cmd, timeout=None): """docker 的默认 runner：subprocess.run，支持 timeout。""" return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) class CloudExecutor(Executor): """云端执行器：可插拔的通用 HTTP runner 接口（厂商无关）。 本轮不假装接通某个云厂商（ECS、Cloud Run 等专属适配明确 不在范围）。它只做一件事：把任务 POST 到一个配置好的通用 runner endpoint，endpoint 返回任务状态，本类据此返回结果。 配置（只走参数或环境变量，不写文件）： - endpoint：显式参数 > CLOUD_RUNNER_ENDPOINT；未设时 execute 一律 raise RuntimeError，绝不静默降级到 Docker/Local。 - 认证：auth_header 参数或 CLOUD_RUNNER_AUTH_HEADER 给 header 名；token 值只从环境变量取（CLOUD_RUNNER_AUTH_TOKEN_ENV 指定的变量名，默认 CLOUD_RUNNER_AUTH_TOKEN），只放进请求头， 绝不写进日志、错误文本或返回结果。 - 请求体只有 id、prompt、repo_url（task["repo"]），不含任何密钥。 - http_client 供单测注入：http_client(url, payload, headers) 返回带 status 字段的 dict（或具 .status/.json 的对象）； 真实路径走 urllib.request。 - dispatch_task/run_task：worker 的 run_one 走 dispatch_task， 这里直接委托给 execute，保证 cloud worker 池能真正派发。 """ def __init__(self, endpoint=None, http_client=None, auth_header=None, auth_token_env=None, env=None): self._env = env if env is not None else os.environ self.endpoint = endpoint or self._env.get("CLOUD_RUNNER_ENDPOINT") self.http_client = http_client self.auth_header = auth_header or self._env.get( "CLOUD_RUNNER_AUTH_HEADER") self.auth_token_env = auth_token_env or self._env.get( "CLOUD_RUNNER_AUTH_TOKEN_ENV") or "CLOUD_RUNNER_AUTH_TOKEN" def _headers(self): headers = {"Content-Type": "application/json"} if self.auth_header: try: token = self._env.get(self.auth_token_env) except Exception: token = None if token: headers[self.auth_header] = token return headers def _post(self, payload): if self.http_client is not None: return self.http_client(self.endpoint, payload, self._headers()) import urllib.request req = urllib.request.Request( self.endpoint, data=json.dumps(payload).encode("utf-8"), headers=self._headers(), method="POST") with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read().decode("utf-8")) def execute(self, task) -> dict: if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") if not self.endpoint: raise RuntimeError( "cloud runner not configured: set CLOUD_RUNNER_ENDPOINT " "to a generic runner endpoint; no cloud account/runner " "configured, refusing to silently fall back to " "Docker/Local") payload = { "id": task["id"], "prompt": task.get("prompt") or "", "repo_url": task.get("repo"), } # 任务带 model 时（worker 每次都设，fallback 时是新模型） # 才发 model 字段：runner 若认就按它跑，不认则忽略；无模型 # 的任务请求体与以前完全一致，不含任何密钥。 if task.get("model"): payload["model"] = task["model"] result = self._post(payload) if isinstance(result, dict): status = result.get("status") stdout = result.get("stdout", "") else: status = getattr(result, "status", None) try: stdout = result.json().get("stdout", "") except Exception: stdout = "" if status == "completed": final = "completed" else: final = "failed" return { "executor": "cloud", "task_id": task["id"], "status": final, "stdout": _truncate_stdout(stdout), "returncode": 0 if final == "completed" else 1, "host_status": None, } def dispatch_task(self, task) -> dict: return self.execute(task) def run_task(self, task) -> dict: return self.execute(task)