diff --git a/CHANGELOG.md b/CHANGELOG.md index bb9dba6..b338848 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,26 @@ 语义化版本见 `VERSION`（当前 0.1.0）。以下按日期倒序，只列已交付 的大项，每项附实测证据；未建成的东西在“诚实边界”里明说，不夸大。 +## 2026-10-26：收尾（Docker 沙箱实测 + 云端合并状态） + +- Docker 沙箱执行器接通并在外部主机上真实跑通：镜像 + `codex-app-server:latest`（codex 0.164.0-alpha.9，codex-resources + 必须放在 codex 二进制旁边、且 bwrap 进 PATH，否则容器内报 + bubblewrap is unavailable）、worker `--executor docker`（额外 + 挂载 `/app/work` 与 `/var/lib/codex-app-server`，容器用户取 + 宿主服务用户 uid/gid）、Docker 任务 triple 加函数约 18 秒完成、 + 容器内合并后主仓库含 add + mul2 + triple、容器经 --rm 清理。 + 诚实边界：这台主机/内核禁非特权 user namespace，bwrap 在 + 主机和容器内都起不来，因此生产模式是 Docker 隔离 + 容器内 + danger-full-access（host-config-docker.json），不是双层沙箱。 +- 云端合并状态：新增 `POST /tasks/{id}/merge`，只同步状态、 + 不执行合并（合并仍由 host.py merge 执行，成功后再调本接口）； + 仅 completed 可改为 merged，maintainer 以上（approve_merge）， + 其它状态与重复调用返回 409，越权 403，审计记 merge_task。 +- 实测证据：云端测试 219 个、host 测试 27 个、合并状态单测 + 4 个，全部通过（`python3 -m unittest discover -s cloud/tests` + 与 `python3 -m unittest discover -s tests`）。 + ## 2026-10-26：缺口补齐（云端控制面第二批） - 记忆 / KB / 成本 / 模型接成 HTTP 端点：记忆写/查/ supersede、 diff --git a/cloud/README.md b/cloud/README.md index 8e18bf1..3f30561 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -42,9 +42,11 @@ queued → running → completed / failed / interrupted / merged / merge-failed ``` 本版 server 只流转 queued → running（心跳）→ interrupted（取消 / -sweep 回收）。`completed` / `merged` 由 worker / 控制面后续调用方在 -跑完模型回合、合完分支后回写，本 server 不提供“跑模型”端点，也不 -要假装一次 POST 就能让任务变 completed。 +sweep 回收），以及 completed → merged（见下表 `POST /tasks/{id}/merge`）。 +`completed` 由 worker 在跑完模型回合后回写；`merged` 只由合并 +状态接口同步，不执行合并本身——合并仍由 host.py merge（本地 +或 Docker 容器内）执行，成功后再调本接口，云端不该在任务 +completed 时自动改主分支。 ## 启动方式 @@ -166,6 +168,7 @@ psycopg，写明“需安装 psycopg”），绝不静默回落到 SQLite， | `GET /tasks?project_id=...` | 该项目 `read` | 返回该项目任务列表：`{"tasks": [...]}`；无权限 `403` | | `GET /tasks/{id}` | 该任务所属项目 `read` | 无权限 `403`；不存在 `404` | | `POST /tasks/{id}/cancel` | 该项目 `create_task` | `queued` / `running` 改为 `interrupted` 并记审计 `cancel_task`；已是终态（completed / failed / interrupted / merged / merge-failed）返回 `400`；不存在 `404` | +| `POST /tasks/{id}/merge` | 该项目 `approve_merge`（maintainer 以上） | 只同步状态，不执行合并：仅 `completed` 可改为 `merged` 并记审计 `merge_task`，其它状态（含已 merged 的重复调用）返回 `409`；不存在 `404`，越权 `403`，无 key `401` | | `POST /tasks/{id}/heartbeat` | 该项目 `create_task` | 更新 `last_heartbeat = now`；状态为 `queued` 则改为 `running`；返回任务 | | `POST /admin/sweep` | 认证即可 | 调 `sweep_stale_tasks(db_path, threshold_seconds=300)`（body 可传 `threshold_seconds` 覆盖默认值）；把 `running` 且 `last_heartbeat`（无心跳则用 `updated_at`）早于 `now - threshold` 的任务改为 `interrupted` 并记审计 `sweep_interrupted`；返回 `{"swept": [ids]}` | | `POST /tasks/{id}/cost` | 该任务所属项目 `read` | 成本聚合：`{task_id, project_id, total_tokens, usd_micros, turns, fallback_count, actual_tokens, estimated_tokens}`；真数与估算分开；不存在 `404`，越权 `403` | @@ -258,9 +261,11 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳 - **Slack App**：需要真实 Slack App（签名密钥、斜杠命令、事件订阅） 才能接通 `SLACK_SIGNING_SECRET` 背后的入站；目前只有验签与解析， 且 `NotificationService` 的 Slack 发送端仍是占位注入。 -- **云容器账号**：`CloudExecutor` 仍是占位（无账号即抛错），Docker - 执行也未接通容器挂载 / 网络隔离 / 资源限额；接通前不要宣称任务 - 已在云端或容器里跑。 +- **云容器账号**：`CloudExecutor` 仍是占位（无账号即抛错）。 + Docker 执行已接通（见下文“Docker 沙箱执行器”），但真实 + docker run、镜像构建、资源限额实测仍需在外部主机上做： + 本 VM 没有 Docker，本轮未连外部主机、未向真实外部服务发 + 请求、未跑真实模型回合，不要宣称 Docker 沙箱已在真实环境验证。 - **模型 key 供给方式**：这是当前最实在的卡点。上次 real-nightly-tests automation 就卡在这里：定时进程手里没有中转站 key，key 从不存文件、只在进程环境里 transient 供给。控制面要跑 @@ -308,3 +313,46 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳 的 Store 连接已统一走 cloud.db 的 WAL + busy_timeout 连接 （Postgres 后端仍是骨架，驱动缺失时启动即失败、不先建 SQLite 文件，这一点迁移后未退化）。 + +## Docker 沙箱执行器（本轮新增） + +- 用法：worker 选 docker 执行器—— + `python3 -m cloud.worker --db <库> --executor docker + [--docker-image codex-app-server:latest] [--docker-cpus 2] + [--docker-memory 2g] [--docker-user 1000:1000] + [--docker-worktrees-root <目录>] [--docker-timeout <秒>]`； + 或在代码里直接传 `DockerExecutor(...)` 给 `Worker(executor=...)`。 + 选 docker 但本机无 docker 时任务明确失败，error_text 含 + docker 缺失，绝不静默降级为本地执行；选 local 时行为不变。 +- 命令构造（示例，任务 id 为 t1、worktree 为 /tmp/wt-t1 时）： + `docker run --rm --name codex-t1 --user 1000:1000 --cpus 2 + --memory 2g -v /tmp/wt-t1:/work -v <代码目录>:/app:ro -w /work + -e XCM_API_KEY codex-app-server:latest python3 /app/host.py + dispatch --name t1 --task <提示>`。要点：只挂任务自己的 + worktree（可写）和代码目录（只读），绝不挂宿主根目录、绝不 + 挂任何含密钥的目录；容器名由任务 id 派生（仅安全字符）； + `-e` 只传 env_key 名，不带值，key 缺失时容器根本不启动； + 资源限额默认 2 核 / 2G，可配置；容器内进程以非 root 用户跑。 +- 镜像：`cloud/docker/Dockerfile`，基于 python:3.12-slim，装 + git，复制调用方提供的配对 codex 二进制（codex + + codex-code-mode-host + codex-resources）。构建命令示例（在 + 仓库根目录，构建上下文就是仓库根）： + `docker build -f cloud/docker/Dockerfile + -t codex-app-server:latest --build-arg CODEX_BIN_DIR=bin .` + bin/ 不进仓库，构建主机上没有配对二进制就构建失败；代码 + 不在镜像里，运行时只读挂载到 /app，代码改了不用重新构建， + codex 版本变了才需要重新构建；镜像里不含任何 key。 +- 清理与取消：`docker run --rm`，正常失败也自动删容器；超时 + 时先 docker kill 再 docker rm -f；worker 取消时按任务派生的 + 容器名 docker kill，不只改数据库状态。 +- 网络白名单的诚实边界：docker run 默认 bridge 网络，本执行 + 器不做 egress 白名单——真正的“只放行模型 API”需要宿主防火 + 墙或代理规则，那是部署项，不是本轮代码假装做到的，不要宣称 + 已实现网络白名单。另有一项待实测：codex 自带的 bubblewrap + 沙箱在容器内是否可用，取决于宿主的容器权限配置，本镜像不 + 假装解决；若 bwrap 不可用，host.py 的 sandbox 参数仍可退到 + read-only / workspace-write 的控制面约束。 +- 仍需在外部主机上实测（本 VM 没有 Docker，本轮未连外部主机）： + 真实 docker build、真实 docker run、资源限额在真实负载下的 + 表现、bwrap 在容器内的可用性、以及取消时容器确实消失。 + diff --git a/cloud/docker/Dockerfile b/cloud/docker/Dockerfile new file mode 100644 index 0000000..87df651 --- /dev/null +++ b/cloud/docker/Dockerfile @@ -0,0 +1,44 @@ +# Codex app-server 任务镜像：只含运行环境 + codex 二进制，代码运行时挂载。 +# +# 构建（在 codex-app-server 仓库根目录，构建上下文就是仓库根）： +# docker build -f cloud/docker/Dockerfile -t codex-app-server:latest \ +# --build-arg CODEX_BIN_DIR=bin . +# +# 说明（不要绕过去）： +# - CODEX_BIN_DIR 是调用方在构建主机上提供的配对二进制目录 +# （npm 安装 @openai/codex 后自带 codex + codex-code-mode-host + +# codex-resources/bwrap，约 367M）。bin/ 不进仓库、不进镜像层 +# 之外的任何地方；没有配对二进制就不要构建，构建会直接失败， +# 绝不静默用一个缺 sibling 的裸二进制。 +# - host.py / cloud/ 代码不在镜像里：运行时由 DockerExecutor 以 +# 只读方式挂载到 /app（-v <代码目录>:/app:ro），代码改了不用 +# 重新构建镜像；codex 版本变了才需要重新构建。 +# - 任务自己的 worktree 运行时挂载到 /work（可写），绝不挂载宿主 +# 根目录、绝不挂载任何含密钥的目录。 +# - 镜像里不含任何 key：模型 key 运行时由 DockerExecutor 以 +# `-e ` 从宿主进程环境注入容器，不写进 Dockerfile、 +# 不写进镜像层、不写进日志。 +# - codex 自带的 bubblewrap 沙箱在容器内是否可用，取决于宿主 +# 的容器权限配置，本镜像不假装解决：真实环境实测时若 bwrap +# 不可用，host.py 的 sandbox 参数仍可退到 read-only / +# workspace-write 的控制面约束，见 cloud/README.md。 +ARG PYTHON_VERSION=3.12-slim +FROM python:${PYTHON_VERSION} +RUN apt-get update \ + && apt-get install -y --no-install-recommends git \ + && rm -rf /var/lib/apt/lists/* +ARG CODEX_BIN_DIR=bin +COPY ${CODEX_BIN_DIR}/codex /usr/local/bin/codex +COPY ${CODEX_BIN_DIR}/codex-code-mode-host /usr/local/bin/codex-code-mode-host +# 关键：codex 找 bundled bwrap 只看自身旁边的 codex-resources， +# 不能放到 /usr/local/lib 下（上一版镜像就是这么丢的，导致 +# 容器内报 bubblewrap is unavailable，工具全失败）。 +# 同时把 bwrap 放进 PATH，双保险。 +COPY ${CODEX_BIN_DIR}/codex-resources /usr/local/bin/codex-resources +RUN chmod +x /usr/local/bin/codex /usr/local/bin/codex-code-mode-host \ + && ln -sf /usr/local/bin/codex-resources/bwrap /usr/local/bin/bwrap \ + && /usr/local/bin/bwrap --version +WORKDIR /work +# 无 ENTRYPOINT：DockerExecutor 显式传完整容器内命令 +# （python3 /app/host.py dispatch/run/status ...），镜像只提供环境。 +CMD ["python3", "--version"] diff --git a/cloud/sandbox.py b/cloud/sandbox.py index c0113f0..8dbafc5 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -5,11 +5,16 @@ - Executor：基类，子类必须实现 execute(task)。 - LocalExecutor：先行实现，复用 host.py 现有 worktree + codex 沙箱， 通过 host.py CLI 子进程做只读查询封装；不直接跑模型。 -- DockerExecutor：占位，未接通容器执行。 +- 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 @@ -210,31 +215,335 @@ class LocalExecutor(Executor): class DockerExecutor(Executor): - """容器执行器（占位，未接通）。 + """容器执行器（已接通，命令构造可测，真实 docker 需外部主机）。 - 本机无 docker（shutil.which("docker") 为 None）时 raise - RuntimeError("docker not available")，绝不静默降级为本地执行。 - 有 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）。 - 后续接线：容器挂载（把任务 worktree 以只读/受限写挂载进容器）、 - 网络隔离（默认无外网，仅放行模型 API 所需出口）、资源限额与 - 容器回收，届时再替换本占位实现。 + 构造函数：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 execute(self, task) -> dict: + 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): + 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 + 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) + + @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): + """构造完整的 docker run 命令（纯函数，可单测断言）。 + + 顺序固定：docker run --rm --name <容器名> --user <用户> + --cpus <核数> --memory <内存> -v :/work + -v <代码目录>:/app:ro -w /work [-e ] + <镜像> <容器内命令...>。-e 只传名字，不带 =值。 + """ + self._check_docker() + self._check_key() + cmd = ["docker", "run", "--rm", + "--name", self.container_name(task), + "--user", self.user, + "--cpus", self.cpus, + "--memory", self.memory, + "-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，同时代码主体仍只读。 + for host_p, cont_p, mode in self.extra_mounts: + cmd += ["-v", "%s:%s:%s" % (host_p, cont_p, mode)] + if self.env_key: + cmd += ["-e", self.env_key] + cmd += [self.image] + cmd += list(inner_cmd) + return cmd + + def _inner_host_command(self, *args): + """容器内 host.py 命令：python3 /app/host.py [args...]。 + + config_path 若已给，容器内用 /app/ 指代它—— + 调用方须保证配置文件就在代码目录里（host.py 配置约定 + 本就放在项目目录）；不在代码目录时不静默猜路径， + 由 _inner_host_command 的调用方在传 config 前自行确认。 + 这里只做最简单的同名映射，不展开宿主绝对路径进容器。 + """ + cmd = ["python3", "/app/host.py"] + if self.config_path is not None: + cmd += ["--config", + "/app/" + os.path.basename(self.config_path)] + cmd += list(args) + return cmd + + # -- runner 调用 / 清理 -------------------------------------------- + def _call_runner(self, cmd): + """调 runner：优先传 timeout，假 runner 只接受 cmd 时回退。""" + try: + return self.runner(cmd, timeout=self.timeout) + except TypeError: + return self.runner(cmd) + + 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): + """在容器里跑一轮内层命令，返回 LocalExecutor 同款结果 dict。 + + 超时：runner 抛 subprocess.TimeoutExpired（或同名异常） + 时先 kill+rm 容器，再返回 failed（stdout 截断 + timeout + 说明，returncode 为 None）；其它 runner 异常也如实返回 + failed，不编造成功。 + """ task_id = task.get("id") if isinstance(task, dict) else None if task_id is None: raise ValueError("任务缺少 id") - if shutil.which("docker") is None: - raise RuntimeError("docker not available") - return { - "executor": "docker", - "task_id": task_id, - "status": "failed", - "stdout": "docker executor not implemented: " - "container mount and network isolation pending", - "returncode": None, - } + cmd = self.build_run_command(task, inner_cmd) + try: + result = self._call_runner(cmd) + 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) -> 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"]) + repo = task.get("repo") + if repo: + inner += ["--repo", repo] + return self._run_in_container(task, inner) + + def run_task(self, task) -> 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"]) + return self._run_in_container(task, inner) + + 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): diff --git a/cloud/server.py b/cloud/server.py index 0dc0a00..09daed0 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -14,8 +14,8 @@ cost 的 CostStore、webhooks 的 NotificationService）本版尚未接成 HTTP 端点，留给 worker 接线时调用；本 server 只接入其中两处： AuthStore（认证/权限/审计）与 webhooks 的验签/解析函数。 -- DockerExecutor/CloudExecutor 目前仍是占位（见 sandbox.py）， - 本 server 不假装已接通容器或云端执行。 +- DockerExecutor 已接通（见 sandbox.py），CloudExecutor 目前仍是占位 + （见 sandbox.py），本 server 不假装已接通云端执行。 导入规则：只用绝对包导入（from cloud.auth import ...）。绝不要把 cloud/ 目录加进 sys.path：标准库已有 queue 模块，会和 cloud/queue.py @@ -367,6 +367,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._task_cost(user_id, rest[: -len("/cost")]) elif rest.endswith("/cancel") and method == "POST": self._cancel_task(user_id, rest[: -len("/cancel")]) + elif rest.endswith("/merge") and method == "POST": + self._merge_task(user_id, rest[: -len("/merge")]) elif rest.endswith("/heartbeat") and method == "POST": self._heartbeat_task(user_id, rest[: -len("/heartbeat")]) else: @@ -984,6 +986,53 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): store.audit(user_id, task["project_id"], "cancel_task", task_id) self._send_json(200, task) + def _merge_task(self, user_id, task_id): + """把云端任务标记为已合并：只补状态，不执行合并本身。 + + 合并本身仍由 host.py merge（本地或 Docker 容器内）执行， + 成功后再调本接口同步云端状态——两者分离是故意的：合并 + 是要人看 diff 才点的操作，云端不该在任务 completed 时 + 自动把主分支改了。规则：任务必须存在（404）、调用方 + 必须是项目 maintainer 以上（approve_merge，403）、任务 + 必须是 completed（其它状态一律 409，含已 merged 的重复 + 调用，避免二次合并混淆）。成功后写审计 merge_task。 + """ + row = self._fetch_task(task_id) + if row is None: + self._send_json(404, {"error": "task not found"}) + return + task = _task_to_dict(row) + with _auth_store(self.server.db_path) as store: + if not store.check_permission( + user_id, task["project_id"], "approve_merge"): + self._send_json(403, {"error": "forbidden"}) + return + if task["status"] != "completed": + self._send_json(409, { + "error": "only completed tasks can be marked merged", + "status": task["status"], + }) + return + now = time.time() + conn = self._db() + try: + conn.execute( + "UPDATE tasks SET status = 'merged', updated_at = ?" + " WHERE id = ?", + (now, task_id), + ) + conn.commit() + task = _task_to_dict( + conn.execute( + "SELECT * FROM tasks WHERE id = ?", (task_id,) + ).fetchone() + ) + finally: + conn.close() + with _auth_store(self.server.db_path) as store: + store.audit(user_id, task["project_id"], "merge_task", task_id) + self._send_json(200, task) + def _heartbeat_task(self, user_id, task_id): row = self._fetch_task(task_id) if row is None: diff --git a/cloud/tests/test_docker_sandbox.py b/cloud/tests/test_docker_sandbox.py new file mode 100644 index 0000000..2470a71 --- /dev/null +++ b/cloud/tests/test_docker_sandbox.py @@ -0,0 +1,339 @@ +"""Docker 沙箱执行器单测：命令构造、安全挂载、超时与取消。 + +运行（在 ~/workspace/codex-app-server 下）： + python3 -m unittest discover -s cloud/tests -p "test_docker_sandbox.py" -v + +说明：全部用注入的假 docker runner，不调用真实 docker、不联网、 +不跑真实模型回合；env 映射里的值只是占位假名，绝不是真实密钥。 +""" + +import json +import os +import sqlite3 +import subprocess +import sys +import tempfile +import time +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +try: + from cloud.sandbox import DockerExecutor + from cloud.server import init_db + from cloud.worker import Worker +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.sandbox import DockerExecutor + from cloud.server import init_db + from cloud.worker import Worker + + +def _ok_runner_factory(seen): + def fake(cmd, timeout=None): + seen.append(list(cmd)) + return SimpleNamespace( + returncode=0, + stdout=json.dumps({"status": "completed", "text": "done"}), + stderr="") + return fake + + +def _insert_task(db_path, task_id, prompt="do work"): + now = time.time() + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " idempotency_key, depends_on, priority, created_by," + " created_at, updated_at, last_heartbeat)" + " VALUES(?, 'p1', ?, 'queued', NULL, '[]', 0, 'tester', ?, ?, NULL)", + (task_id, prompt, now, now), + ) + conn.commit() + finally: + conn.close() + + +def _task_status(db_path, task_id): + conn = sqlite3.connect(db_path) + try: + row = conn.execute( + "SELECT status, error_text FROM tasks WHERE id = ?", + (task_id,)).fetchone() + finally: + conn.close() + return row + + +class DockerCommandTestCase(unittest.TestCase): + def _executor(self, seen, **kwargs): + kwargs.setdefault("runner", _ok_runner_factory(seen)) + kwargs.setdefault("env", {"XCM_API_KEY": "placeholder-not-a-real-key"}) + return DockerExecutor("/fake/host.py", **kwargs) + + def test_command_mounts_only_task_worktree_and_ro_code(self): + seen = [] + with mock.patch("cloud.sandbox.shutil.which", + return_value="/usr/bin/docker"): + ex = self._executor(seen) + cmd = ex.build_run_command( + {"id": "t1", "worktree": "/tmp/wt-t1"}, + ["python3", "/app/host.py", "status", "--task", "t1"]) + mounts = [cmd[i + 1] for i, a in enumerate(cmd) if a == "-v"] + self.assertEqual(mounts, ["/tmp/wt-t1:/work", "/fake:/app:ro"]) + # 绝不挂载宿主根目录。 + self.assertNotIn("/:/work", mounts) + self.assertNotIn("/:/app:ro", mounts) + self.assertIn("--rm", cmd) + self.assertIn("/tmp/wt-t1:/work", cmd) + + def test_command_resource_limits_and_non_root_user(self): + seen = [] + with mock.patch("cloud.sandbox.shutil.which", + return_value="/usr/bin/docker"): + ex = self._executor(seen, cpus="1", memory="512m", + user="1000:1000") + cmd = ex.build_run_command( + {"id": "t1", "worktree": "/tmp/wt"}, + ["echo", "hi"]) + self.assertEqual(cmd[cmd.index("--cpus") + 1], "1") + self.assertEqual(cmd[cmd.index("--memory") + 1], "512m") + self.assertEqual(cmd[cmd.index("--user") + 1], "1000:1000") + self.assertNotEqual(cmd[cmd.index("--user") + 1], "0") + self.assertNotEqual(cmd[cmd.index("--user") + 1], "root") + + def test_command_defaults_are_2cpu_2g(self): + seen = [] + with mock.patch("cloud.sandbox.shutil.which", + return_value="/usr/bin/docker"): + ex = self._executor(seen) + cmd = ex.build_run_command( + {"id": "t1", "worktree": "/tmp/wt"}, ["echo", "hi"]) + self.assertEqual(cmd[cmd.index("--cpus") + 1], "2") + self.assertEqual(cmd[cmd.index("--memory") + 1], "2g") + + def test_container_name_sanitized(self): + seen = [] + with mock.patch("cloud.sandbox.shutil.which", + return_value="/usr/bin/docker"): + ex = self._executor(seen) + self.assertEqual(ex.container_name({"id": "a b/c"}), + "codex-a-b-c") + cmd = ex.build_run_command( + {"id": "a b/c", "worktree": "/tmp/wt"}, ["echo", "hi"]) + self.assertIn("codex-a-b-c", cmd) + self.assertEqual(cmd[cmd.index("--name") + 1], "codex-a-b-c") + + def test_env_name_only_no_value_in_command(self): + seen = [] + with mock.patch("cloud.sandbox.shutil.which", + return_value="/usr/bin/docker"): + ex = self._executor(seen) + cmd = ex.build_run_command( + {"id": "t1", "worktree": "/tmp/wt"}, ["echo", "hi"]) + joined = " ".join(cmd) + self.assertIn("-e", cmd) + self.assertIn("XCM_API_KEY", cmd) + # 值绝不出现在命令列表里（-e 只传名，docker 从宿主环境取值）。 + self.assertNotIn("placeholder-not-a-real-key", joined) + self.assertNotIn("placeholder-not-a-real-key=", joined) + + def test_missing_key_does_not_start_container(self): + seen = [] + with mock.patch("cloud.sandbox.shutil.which", + return_value="/usr/bin/docker"): + ex = DockerExecutor("/fake/host.py", + runner=_ok_runner_factory(seen), env={}) + with self.assertRaises(RuntimeError) as ctx: + ex.dispatch_task({"id": "t1", "prompt": "hi", + "worktree": "/tmp/wt"}) + self.assertIn("Missing environment variable", str(ctx.exception)) + self.assertEqual(seen, []) + + def test_missing_docker_does_not_call_runner(self): + seen = [] + with mock.patch("cloud.sandbox.shutil.which", return_value=None): + ex = self._executor(seen) + with self.assertRaises(RuntimeError) as ctx: + ex.dispatch_task({"id": "t1", "prompt": "hi", + "worktree": "/tmp/wt"}) + self.assertIn("docker not available", str(ctx.exception)) + self.assertEqual(seen, []) + + def test_dispatch_and_run_shapes(self): + seen = [] + with mock.patch("cloud.sandbox.shutil.which", + return_value="/usr/bin/docker"): + ex = self._executor(seen) + dispatched = ex.dispatch_task( + {"id": "t1", "prompt": "do it", "worktree": "/tmp/wt"}) + ran = ex.run_task( + {"id": "t1", "prompt": "again", "worktree": "/tmp/wt"}) + self.assertEqual(dispatched["status"], "completed") + self.assertEqual(dispatched["executor"], "docker") + dispatch_cmd = seen[0] + self.assertIn("dispatch", dispatch_cmd) + self.assertIn("t1", dispatch_cmd) + self.assertIn("do it", dispatch_cmd) + run_cmd = seen[1] + self.assertIn("run", run_cmd) + self.assertIn("again", run_cmd) + + def test_execute_is_readonly_status_not_dispatch(self): + seen = [] + with mock.patch("cloud.sandbox.shutil.which", + return_value="/usr/bin/docker"): + ex = self._executor(seen) + result = ex.execute({"id": "t1", "worktree": "/tmp/wt"}) + self.assertEqual(result["status"], "completed") + self.assertIn("status", seen[0]) + self.assertNotIn("dispatch", seen[0]) + # run 在命令里会出现 "--name" 等，但内层命令不应是 run： + # 内层 host.py 命令的第一个动词是 status。 + inner = seen[0][seen[0].index("codex-app-server:latest") + 1:] + self.assertEqual(inner[2], "status") + + def test_timeout_kills_and_removes_container(self): + seen = [] + + def fake(cmd, timeout=None): + seen.append(list(cmd)) + if len(cmd) > 1 and cmd[1] == "run": + raise subprocess.TimeoutExpired(cmd, timeout) + return SimpleNamespace(returncode=0, stdout="", stderr="") + + with mock.patch("cloud.sandbox.shutil.which", + return_value="/usr/bin/docker"): + ex = DockerExecutor("/fake/host.py", runner=fake, + env={"XCM_API_KEY": "placeholder"}, + timeout=5) + result = ex.dispatch_task( + {"id": "t9", "prompt": "slow", "worktree": "/tmp/wt"}) + self.assertEqual(result["status"], "failed") + self.assertIn("timeout", result["stdout"]) + kinds = [(c[1], c[2]) for c in seen if len(c) > 2] + self.assertIn(("kill", "codex-t9"), kinds) + self.assertIn(("rm", "-f"), [(c[1], c[2]) for c in seen + if len(c) > 2]) + + def test_kill_container_calls_kill_and_rm(self): + seen = [] + with mock.patch("cloud.sandbox.shutil.which", + return_value="/usr/bin/docker"): + ex = self._executor(seen) + ok = ex.kill_container({"id": "t2"}) + self.assertTrue(ok) + self.assertEqual([c[1] for c in seen], ["kill", "rm"]) + self.assertEqual(seen[0][2], "codex-t2") + self.assertEqual(seen[1][2], "-f") + self.assertEqual(seen[1][3], "codex-t2") + + +class WorkerDockerTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + self.env = {"XCM_API_KEY": "placeholder-not-a-real-key"} + + def tearDown(self): + self.tmp.cleanup() + + def test_worker_docker_missing_fails_task_with_docker_error(self): + seen = [] + + def fake(cmd, timeout=None): + seen.append(list(cmd)) + return SimpleNamespace(returncode=0, stdout="{}", stderr="") + + _insert_task(self.db_path, "d1", prompt="do it") + with mock.patch("cloud.sandbox.shutil.which", return_value=None): + ex = DockerExecutor("/fake/host.py", runner=fake, env=self.env) + worker = Worker(self.db_path, "/fake/host.py", + env={"XCM_API_KEY": "placeholder-not-a-real-key"}, executor=ex) + result = worker.run_one() + self.assertTrue(result["dispatched"]) + self.assertEqual(result["status"], "failed") + self.assertIn("docker", result["error"]) + self.assertEqual(seen, []) + status, error_text = _task_status(self.db_path, "d1") + self.assertEqual(status, "failed") + self.assertIn("docker", error_text) + + def test_worker_docker_success(self): + seen = [] + _insert_task(self.db_path, "d2", prompt="do it") + with mock.patch("cloud.sandbox.shutil.which", + return_value="/usr/bin/docker"): + ex = DockerExecutor("/fake/host.py", + runner=_ok_runner_factory(seen), + env=self.env) + worker = Worker(self.db_path, "/fake/host.py", + env={"XCM_API_KEY": "placeholder-not-a-real-key"}, executor=ex) + result = worker.run_one() + self.assertTrue(result["dispatched"]) + self.assertEqual(result["status"], "completed") + self.assertEqual(result["executor"], "docker") + self.assertTrue(any(c[1] == "run" for c in seen)) + + def test_worker_cancel_kills_docker_container(self): + seen = [] + _insert_task(self.db_path, "d3", prompt="do it") + with mock.patch("cloud.sandbox.shutil.which", + return_value="/usr/bin/docker"): + ex = DockerExecutor("/fake/host.py", + runner=_ok_runner_factory(seen), + env=self.env) + worker = Worker(self.db_path, "/fake/host.py", + env={"XCM_API_KEY": "placeholder-not-a-real-key"}, executor=ex) + # 模拟 dispatch 阻塞期间：worker 已把容器名记下来。 + worker._running_task_id = "d3" + worker._running_container = "codex-d3" + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "UPDATE tasks SET status = 'running' WHERE id = 'd3'") + conn.commit() + finally: + conn.close() + self.assertTrue(worker.cancel("d3")) + self.assertEqual([c[1] for c in seen], ["kill", "rm"]) + status, _ = _task_status(self.db_path, "d3") + self.assertEqual(status, "interrupted") + + +if __name__ == "__main__": + unittest.main() + +class TestDockerExtraMounts(DockerCommandTestCase): + def test_extra_mounts_appended_after_base_mounts(self): + with tempfile.TemporaryDirectory() as td: + work = os.path.join(td, "work") + os.makedirs(work) + state = os.path.join(td, "host-state.json") + open(state, "w").write("{}") + with mock.patch("cloud.sandbox.shutil.which", + return_value="/usr/bin/docker"): + ex = DockerExecutor("/fake/host.py", runner=_ok_runner_factory([]), + env={"XCM_API_KEY": "placeholder-not-a-real-key"}, + extra_mounts=[f"{work}:/app/work", f"{state}:/app/host-state.json"]) + cmd = ex.build_run_command({"id": "t1"}, ["python3", "/app/host.py", "status"]) + # 基础挂载在前，额外挂载在后，且 /app/work 覆盖只读子路径 + self.assertIn(f"{work}:/app/work:rw", cmd) + self.assertIn(f"{state}:/app/host-state.json:rw", cmd) + mounts = [cmd[i + 1] for i, a in enumerate(cmd) if a == "-v"] + self.assertEqual(mounts[1], "/fake:/app:ro") + self.assertIn(f"{work}:/app/work:rw", mounts) + self.assertLess(mounts.index("/fake:/app:ro"), mounts.index(f"{work}:/app/work:rw")) + + def test_extra_mounts_reject_dangerous_paths(self): + with self.assertRaises(ValueError): + DockerExecutor("/fake/host.py", extra_mounts=["/: /x"]) + with self.assertRaises(ValueError): + DockerExecutor("/fake/host.py", extra_mounts=["/etc/codex-app-server/env:/x"]) + with self.assertRaises(ValueError): + DockerExecutor("/fake/host.py", extra_mounts=["/tmp:/app"]) + with self.assertRaises(ValueError): + DockerExecutor("/fake/host.py", extra_mounts=["/nonexistent-xyz:/x"]) diff --git a/cloud/tests/test_queue_sandbox.py b/cloud/tests/test_queue_sandbox.py index 5cb84d7..fe6d082 100644 --- a/cloud/tests/test_queue_sandbox.py +++ b/cloud/tests/test_queue_sandbox.py @@ -189,12 +189,29 @@ class TestExecutors(unittest.TestCase): self.assertIn("docker not available", str(ctx.exception)) # 报错而不是返回 local 结果：绝不静默降级为本地执行 - def test_docker_executor_with_docker_is_placeholder_failure(self): - with mock.patch("sandbox.shutil.which", return_value="/usr/bin/docker"): - result = DockerExecutor().execute({"id": "t1"}) + def test_docker_executor_with_docker_runs_container(self): + seen = [] + + def fake_runner(cmd, timeout=None): + seen.append(cmd) + return SimpleNamespace(returncode=0, stdout="ok-status") + + with mock.patch("sandbox.shutil.which", + return_value="/usr/bin/docker"): + executor = DockerExecutor( + "/fake/host.py", runner=fake_runner, + env={"XCM_API_KEY": "placeholder-not-a-real-key"}) + result = executor.execute({"id": "t1", "worktree": "/tmp/wt1"}) self.assertEqual(result["executor"], "docker") - self.assertEqual(result["status"], "failed") - self.assertNotEqual(result["executor"], "local") + self.assertEqual(result["status"], "completed") + self.assertEqual(seen[0][0:2], ["docker", "run"]) + self.assertIn("--rm", seen[0]) + self.assertIn("/tmp/wt1:/work", seen[0]) + self.assertIn("/fake:/app:ro", seen[0]) + # -e 只传名，值绝不出现在命令列表里。 + self.assertIn("XCM_API_KEY", seen[0]) + self.assertNotIn("placeholder-not-a-real-key", seen[0]) + self.assertNotIn("placeholder-not-a-real-key", " ".join(seen[0])) def test_cloud_executor_raises_with_cloud_account(self): with self.assertRaises(RuntimeError) as ctx: diff --git a/cloud/tests/test_server_merge.py b/cloud/tests/test_server_merge.py new file mode 100644 index 0000000..f365dcf --- /dev/null +++ b/cloud/tests/test_server_merge.py @@ -0,0 +1,148 @@ +"""云端合并状态端点测试：POST /tasks/{id}/merge。 + +只测状态同步本身，不执行真实合并：合并仍由 host.py merge +（本地或 Docker 容器内）执行，成功后再调本接口把云端 +任务从 completed 改为 merged。规则：404/401/403/409 +语义、maintainer 以上才能合并、审计里有 merge_task。 +""" + +import http.client +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import unittest +from pathlib import Path + +try: + from cloud.auth import AuthStore + from cloud.server import create_server +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.server import create_server + + +class MergeTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.auth = AuthStore(self.db_path) + self.owner_id, self.owner_key = self.auth.create_user("owner") + self.maint_id, self.maint_key = self.auth.create_user("maint") + self.contrib_id, self.contrib_key = self.auth.create_user("contrib") + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + status, self.project = self._req( + "POST", "/projects", {"name": "demo"}, self.owner_key) + self.assertEqual(status, 201) + self.auth.set_member( + self.project["id"], self.owner_id, self.maint_id, "maintainer") + self.auth.set_member( + self.project["id"], self.owner_id, self.contrib_id, "contributor") + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + + def _req(self, method, path, body=None, key=None): + conn = http.client.HTTPConnection( + "127.0.0.1", self.port, timeout=10) + hdrs = {} + payload = None + if body is not None: + payload = json.dumps(body).encode("utf-8") + hdrs["Content-Type"] = "application/json" + if key is not None: + hdrs["X-API-Key"] = key + try: + conn.request(method, path, body=payload, headers=hdrs) + resp = conn.getresponse() + data = resp.read() + finally: + conn.close() + try: + parsed = json.loads(data.decode("utf-8")) if data else None + except ValueError: + parsed = None + return resp.status, parsed + + def _completed_task(self): + status, task = self._req( + "POST", "/tasks", + {"project_id": self.project["id"], "prompt": "x"}, + self.owner_key) + self.assertEqual(status, 201) + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "UPDATE tasks SET status = 'completed' WHERE id = ?", + (task["id"],)) + conn.commit() + finally: + conn.close() + return task + + def test_merge_completed_by_owner(self): + task = self._completed_task() + status, merged = self._req( + "POST", "/tasks/%s/merge" % task["id"], {}, self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(merged["status"], "merged") + status, got = self._req( + "GET", "/tasks/%s" % task["id"], key=self.owner_key) + self.assertEqual(merged["status"], got["status"]) + audit = self.auth.list_audit(self.project["id"]) + self.assertTrue(any(a["action"] == "merge_task" for a in audit)) + + def test_merge_by_maintainer_ok_contributor_forbidden(self): + task = self._completed_task() + status, body = self._req( + "POST", "/tasks/%s/merge" % task["id"], {}, self.contrib_key) + self.assertEqual(status, 403) + status, merged = self._req( + "POST", "/tasks/%s/merge" % task["id"], {}, self.maint_key) + self.assertEqual(status, 200) + self.assertEqual(merged["status"], "merged") + + def test_merge_non_completed_409_and_repeat_409(self): + status, task = self._req( + "POST", "/tasks", + {"project_id": self.project["id"], "prompt": "x"}, + self.owner_key) + self.assertEqual(status, 201) + status, body = self._req( + "POST", "/tasks/%s/merge" % task["id"], {}, self.owner_key) + self.assertEqual(status, 409) + self.assertEqual(body["status"], "queued") + done = self._completed_task() + status, _ = self._req( + "POST", "/tasks/%s/merge" % done["id"], {}, self.owner_key) + self.assertEqual(status, 200) + status, body = self._req( + "POST", "/tasks/%s/merge" % done["id"], {}, self.owner_key) + self.assertEqual(status, 409) + self.assertEqual(body["status"], "merged") + + def test_merge_missing_and_no_key(self): + status, _ = self._req( + "POST", "/tasks/nope/merge", {}, self.owner_key) + self.assertEqual(status, 404) + task = self._completed_task() + status, _ = self._req("POST", "/tasks/%s/merge" % task["id"], {}) + self.assertEqual(status, 401) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 86afcee..5b212c0 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -48,6 +48,12 @@ interrupted，不用执行结果覆盖它。 可注入：runner（假 CLI，见 sandbox.LocalExecutor）与 env 映射 均可注入，单测不跑真实模型回合、不联网、不登录任何服务。 + +执行器选择：Worker 默认用 LocalExecutor；传入 DockerExecutor +（或 CLI --executor docker）时，派发走容器内 host.py，取消时 +按任务派生的容器名 docker kill，不只改数据库状态；本机无 +docker 时 DockerExecutor 在调 runner 前就明确报错，任务标 +failed 且 error_text 含 docker 缺失，绝不静默降级为本地执行。 """ import argparse @@ -66,7 +72,7 @@ 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.sandbox import LocalExecutor +from cloud.sandbox import DockerExecutor, LocalExecutor from cloud.sandbox import _default_runner as _sandbox_default_runner # 与 server.TERMINAL_STATUSES 对齐：这些状态的任务不再被认领或取消。 @@ -189,6 +195,7 @@ class Worker: self.process_factory = process_factory self._running_proc = None self._running_task_id = None + self._running_container = None self._loop_stop = False self.logger = get_logger("cloud.worker") @@ -419,6 +426,18 @@ class Worker: pass self._running_proc = None self._running_task_id = None + # Docker 取消：按任务派生的容器名真杀，不只改数据库状态。 + # 同步路径（runner 阻塞在 docker run 里）下 _running_proc + # 为空，此时靠 _running_container 杀容器；kill 尽力而为， + # 失败也不影响后续的状态标记。 + if self._running_task_id == task_id and \ + self._running_container is not None and \ + hasattr(self.executor, "kill_container"): + try: + self.executor.kill_container(self._running_container) + except Exception: + pass + self._running_container = None conn = _db_connect(self.db_path) try: cur = conn.execute( @@ -451,6 +470,11 @@ class Worker: return task is not None and task["status"] == "interrupted" # -- 缺 key 门 ------------------------------------------------------ + def _executor_name(self): + """执行器名：DockerExecutor -> docker，其它 -> local。""" + if isinstance(self.executor, DockerExecutor): + return "docker" + return "local" def key_available(self): """只查存在性：名在环境里且值非空即 True；值本身绝不返回。""" try: @@ -557,7 +581,7 @@ class Worker: 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": "local", + "status": "failed", "executor": self._executor_name(), "stdout": "", "error": error, "events": [], "cost_record_id": None} @@ -565,6 +589,15 @@ class Worker: fallback_count = 0 result = None error = None + # Docker 同步路径：dispatch 阻塞在 docker run 里时，取消 + # 线程靠 _running_container 按名杀容器（见 cancel()）。 + if hasattr(self.executor, "container_name"): + try: + self._running_container = \ + self.executor.container_name(task) + self._running_task_id = task_id + except Exception: + self._running_container = None try: if self._use_popen(): result = self._execute_via_process(task) @@ -585,6 +618,12 @@ class Worker: "status": "failed", "stdout": "", "returncode": None, "host_status": None, "stderr": ""} error = str(exc) + # 同步路径执行结束：清掉取消用的容器标记（取消路径里 + # 若已清过，这里再清一次无害）。 + self._running_container = None + if self._running_task_id == task_id and \ + self._running_proc is None: + self._running_task_id = None # 取消优先：执行期间被 cancel 的任务保持 interrupted， # 绝不用执行结果覆盖它（见模块 docstring 的取消说明）。 @@ -599,7 +638,7 @@ class Worker: 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": "local", + "status": "interrupted", "executor": self._executor_name(), "stdout": (result or {}).get("stdout", ""), "error": "cancelled", "events": [], "cost_record_id": None} @@ -649,7 +688,7 @@ class Worker: 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": "local", + "status": final, "executor": self._executor_name(), "stdout": stdout, "error": error, "events": events, "cost_record_id": cost_id} @@ -719,6 +758,21 @@ class Worker: return all_results +def _build_cli_executor(args, host_py): + """CLI 的执行器构造：local 时返回 None（Worker 默认 LocalExecutor）， + docker 时返回 DockerExecutor（env_key 名用 Worker 的解析规则， + 值本身绝不从这里传——DockerExecutor 只查存在性，-e 只传名）。""" + 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) + + def main(argv=None): """worker CLI：默认单次 run_all；--loop 进常驻循环。 @@ -737,6 +791,22 @@ def main(argv=None): parser.add_argument("--model", default=None) parser.add_argument("--max-turns", type=int, default=None) parser.add_argument("--max-tasks", type=int, default=None) + parser.add_argument("--executor", default="local", + choices=["local", "docker"], + help="执行器：local 调 host.py 子进程；docker" + " 调 DockerExecutor（无 docker 时任务" + "明确失败，不静默降级）") + 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("--loop", action="store_true", help="常驻循环，每 --interval 秒跑一轮") parser.add_argument("--interval", type=int, default=300) @@ -750,7 +820,8 @@ def main(argv=None): 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, - max_turns=args.max_turns) + max_turns=args.max_turns, + executor=_build_cli_executor(args, host_py)) # CLI 起 worker 前先建表/迁移：全新库直接起循环不应因缺表而炸。 try: from cloud.server import init_db as _init_db