diff --git a/.github/workflows/host-verify.yml b/.github/workflows/host-verify.yml new file mode 100644 index 0000000..246f3fd --- /dev/null +++ b/.github/workflows/host-verify.yml @@ -0,0 +1,14 @@ +name: host-verify +on: + pull_request: + push: + branches: [main] +jobs: + verify: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - run: python3 -m unittest discover -s tests -v diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..bb9dba6 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,227 @@ +# CHANGELOG + +语义化版本见 `VERSION`（当前 0.1.0）。以下按日期倒序，只列已交付 +的大项，每项附实测证据；未建成的东西在“诚实边界”里明说，不夸大。 + +## 2026-10-26：缺口补齐（云端控制面第二批） + +- 记忆 / KB / 成本 / 模型接成 HTTP 端点：记忆写/查/ supersede、 + KB 索引/搜索（按项目隔离）、按任务与按项目/按日的成本（实际与 + 估算分开）、模型总览；全部走 `X-API-Key` + 项目 RBAC， + 401 / 403 / 404 语义与跨项目隔离均有测试。 +- SQLite 并发：共享 db 助手（WAL + busy_timeout + 有界重试）， + 并注明单机上限，Postgres 是下一阶段，不是本版已做。 +- OAuth 脚手架：`cloud/oauth.py`，state 10 分钟过期、生成授权 URL、 + 回调验 state 后用可注入的 HTTP 客户端换 code；secret 只走环境变量， + 没有假登录页，真实 OAuth 应用注册仍缺。 +- 推送订阅与分发：`cloud/push.py`，webhook / slack / webpush / + apns / fcm 通道，即时/摘要/静默偏好，只发状态变更事件；缺发送方/ + 凭证时报 error，绝不静默报成功；真实推送凭证未配、真实推送未发。 +- 实测证据：云端测试从 102 个涨到 157 个，全部通过 + （`python3 -m unittest discover -s cloud/tests`）； + host 测试同期 27 个，全部通过 + （`python3 -m unittest discover -s tests`）； + 另有一次 live smoke：记忆写 201/读通、KB 索引/搜索命中、 + OAuth 登录 URL 只返回不跳转、推送分发报“无发送方”错误、 + 无 key 查项目返回 401。 + +## 2026-10-26：云端控制面（第一批） + +- 新 `cloud/` 包，零第三方依赖、只用 Python 标准库： + `server.py`（ThreadingHTTPServer + SQLite 状态，任务状态机与 + host.py 对齐，含心跳与 stale 回收）、`auth.py`（API key 只存 + SHA-256 哈希、按项目 RBAC owner / maintainer / contributor / + viewer、追加式审计日志）、`queue.py`（依赖拓扑 + 优先级 + + 预算门）、`sandbox.py`（Local / Docker / Cloud 三种执行器， + 缺 Docker 或缺云账号时明确报错、不静默降级）、`memory.py` + （线程摘要，可删/可 supersede，只取 top-k）、`kb.py` + （SQLite FTS，LIKE 兜底，按项目隔离）、`models.py` + （404 / 429 / 5xx / 超时才 fallback，不对 401 / 400 fallback， + 实际用到的模型如实记录）、`cost.py`（读不到账单时返回 + available=false，估算值明确标 estimated，实际与估算分开）、 + `webhooks.py`（GitHub HMAC + Slack 签名与重放窗口，只解析， + 只在状态变更时推送）。 +- 实测证据：云端测试 102 个、host 测试 27 个，全部通过； + smoke：health 正常、无 key 查项目返回 401；代码里无密钥。 +- 诚实边界：这是本地可跑的控制面演示，不是已建成的真实云服务； + 仍缺常驻宿主、GitHub 远端 + PAT、Slack App、云容器账号、 + 定时/worker 跑模型时的模型 key 供给方式、真实 OAuth 应用注册、 + 真实推送凭证；server 本身不跑模型回合。 + +## 2026-10-26：GitHub PR / CI 一条线 + +- `host.py github-workflow`：生成 `.github/workflows/host-verify.yml` + （checkout + python + unittest），幂等。 +- `host.py pr`：校验任务已完成且有分支，push 分支 + `gh pr create`， + 成功后把 pr_url / pr_number 写回任务记录。 +- `host.py pr-checks`：解析 `gh pr checks`，all_pass 仅在 checks + 非空且全部 pass 时为真，空 checks 不算过。 +- `host.py pr-merge`：先查 checks，未全过拒绝且不调 merge， + 全过才 `gh pr merge --merge` 并标 merged，不动本地主分支 + （与本地 `merge` 是两条线）。 +- 实测证据：全部 gh 调用走可 mock 的 run_gh；测试从 19 个涨到 + 27 个，全部通过，含本地 bare 远端 push + 假 gh PR（写回编号 42） + 与 workflow 幂等。 +- 诚实边界：真实 PR 本轮没跑——本仓没有 git remote，gh 未登录； + 真实 PR 仍缺远端与经 Secure Vault 的 PAT，那两样齐了才能跑。 + +## 2026-10-26：第二阶段（daemon / health / 自动恢复） + +- `host.py daemon`：双 fork 到 PID 1，pidfile + host-daemon.log， + 每轮跑 recover + run-due（各为独立 host.py 子进程）； + `daemon-status` / `daemon-stop` 配套。 +- `host.py health`：不跑模型，只查二进制、配置、state 有效性、 + 备份、daemon pid、server initialize + list、缺失 worktree； + 失败时 exit 1。 +- `host.py restore-state`：state 损坏/丢失时从 .bak / .bak.1 / + .bak.2 恢复，损坏文件改名为 .corrupt.<时间戳> 留证。 +- `host.py recover`：超过 `--older-than`（默认 3600 秒）未更新的 + running 任务标 interrupted 并记日志，可用 `run` 续跑； + 故意不自动重跑模型回合，避免未经批准花 relay 额度。 +- 附带修掉一个静默丢数据洞：旧 state 读取把 JSON 解析错误 + 当空 state 吞掉。 +- 实测证据：daemon pid 的 PPID 为 1，测试中每 5 秒 tick、 + stop 后 pidfile 删除；health 先 ok=true（7 个 server 线程）， + 损坏 state 后 ok=false / exit 1，restore 后 ok=true（备份版本 9）； + recover 把 2 小时未动的 running 标 interrupted、1 分钟的保留； + 测试从 15 个涨到 19 个，全部通过。 +- 诚实边界：daemon 本身活不过平台 VM 回收（与微信桥 serve 同理）； + VM 上更稳的定时是 Muse cron 调 run-due / health，不是只靠 daemon + 或 crontab；本轮没有自动装 Muse cron 或 crontab，因为 run-due + 到期会花模型额度，装定时需用户明确指定间隔与是否允许自动跑模型。 + +## 2026-10-26：第二类缺口（全部补齐） + +- 合并自动验测 + 回滚：`merge` 默认仓库有 tests/ 就跑 unittest， + 验测失败 reset 回合并前 HEAD，分支/worktree 保留； + `--no-verify` 跳过，`--verify` 可指定命令。 +- 任务依赖 + 优先级 + 自动重试：jobs 里 depends_on / priority， + 拓扑排序 + 按波次跑，依赖失败时依赖任务标 skipped 不跑； + 依赖任务从依赖分支建 worktree（有血缘关系），不再文件拷贝； + dispatch / parallel 里 `--max-retries`，失败后同 prompt 再跑， + 重试前查预算，默认 0、行为不变。 +- 评审评论回流：`review-comment --file --line --comment` 把评论 + 存进任务，`review` 列出 open 评论，`fix-review` 按 open 评论 + 逐条修、修完标 resolved。 +- server 崩溃重连：Server.ensure_alive，死了重拉并 resume thread + （thread 在 CODEX_HOME 里不丢）；request 遇 BrokenPipe 重试一次； + 单测 kill 进程后重拉 + list 能通。 +- 成本预算 + 失败通知：配置 max_daily_turns，按 turn 数算 + （协议不暴露真实 token 数，这是诚实边界），超预算不派发/ + 不重试；usage 记在 state 里，`doctor` 可看；失败时写 + host-run.log + 跑配置的 notify_command（本地 shell， + 模板 {task} {status}）；`failures` 列失败任务，`doctor` 列 + 失败/缺失 worktree/open 评论。 +- 多机器 state 版本：state 加 version / last_writer / updated_at， + 保存时磁盘版本比加载时新就记 STATE-CONFLICT 日志并合并保留 + 双方任务，不静默丢；真多机中心同步仍是检测、不是云同步。 +- 合并时不再提交缓存文件：merge 前清 worktree 的 + __pycache__ / *.pyc，主仓的 .pyc 改动先还原；单测覆盖 pyc 不入库。 +- 实测证据：测试从 3 个涨到 15 个，全部通过；模型验证在 + gap2 / gap3 仓上跑通（依赖并行、评审评论修复、带自动验测的合并）， + 并在实测中拦下两处真问题：文件拷贝依赖分支导致重复加同一段的 + 合并冲突（-X ours 兜底的新增行核对拦下一次 mul2 丢失并回滚， + 后改从依赖分支建 worktree 修复）、merge 的 git add -A 提交 + __pycache__ / *.pyc（已修）。 +- 诚实边界：独立任务在同一文件位置追加仍会合并冲突，这是 git + 自身行为，host abort / 回滚、不猜合并结果；多机共享 state + 现在能检测、不静默丢，但 ChatGPT 云端式中心同步不在本地 host + 范围内。 + +## 2026-10-26：生产加固 + +- state 文件锁（fcntl，只锁短读/写；保存时按任务/自动化名合并 + 磁盘状态，多进程并发写不丢对方任务；双进程 0.5 秒窗口竞写实测 + 双方任务都保留）。 +- 项目配置 host-config.json（`init` / `config` 命令，CLI > 配置 > + 默认；API key 绝不进配置文件，只走环境变量）。 +- 定时自动化：run-due 在 host-run.log 记派发/跳过， + `automations --cron-line` 只打印 crontab 行不装机， + `--install-cron` 幂等装机；本轮测试里没有实际装 crontab， + 避免未经批准的模型花费，且 VM crontab 可能活不过平台回收， + Muse cron 调 run-due 是更稳的 VM 选项。 +- 诚实边界：锁只防同一台机器，不防两台机器共享同一 state 文件； + host.py 本身不常驻，仍需外部调度器触发 run-due。 + +## 2026-10-26：真实项目跑一轮（在打包前） + +- 在 codex-app-server 本仓（非敏感的 wechat-bridge）上 dogfood： + 先 git init（初始提交，.gitignore 排除 367M bin/、work/、 + host-state*.json、host-run.log、本地 host 配置），真实任务用 + 单独的 host-state-real.json。 +- 并行真实任务：real-tests 加 tests/test_host.py（3 个 unittest， + 全部通过，host.py 本身未改）；real-backup 加 save_state 备份轮转 + （.bak / .bak.1 / .bak.2，在锁内）。 +- 只读委派评审 real-backup 后发现多步轮转非原子、有截断拷贝风险、 + 无 fsync、fcntl-less 平台问题；便宜的修复已在 fix 回合里落地 + （with open 读 state / 写临时文件、修掉 real-tests 发现的 + ResourceWarning、备份经临时文件 + os.replace 拷贝）。 +- 两任务都合并进 main，main 测试全过且无警告，手动双保存确认 + .bak 里是上一版。 +- 真实项目才有的教训：非 git 项目要先 git init 才能用 + worktree / 合并门；缺 .gitignore 会把 bin/ 提交进去； + 真实任务与玩具任务必须用不同的 state 文件。 +- 记录在 TEST-REPORT.md；本轮打包前仍欠的下一步是：安装文档、 + host 快速开始、版本/变更日志、合并后主仓自动验测（那一步当时 + 仍是手动）——其中合并后自动验测已在“第二类缺口”里补上， + 安装文档与版本/变更日志由本 CHANGELOG 与 install.sh / README + 快速开始补上。 + +## 2026-10-25：host 编排（多线程并行、worktree 隔离、评审/合并门） + +- 单个 app-server 进程托管多 thread，后台读线程按 threadId 分发 + turn / completed，多 thread 的 turn 真正并发，不是轮流排队。 +- `dispatch` 自动 `git worktree add -b host/`，一任务一 + worktree；非 git 仓库退化为独立目录拷贝。 +- `review`：status + diff + untracked；`merge`：先在 worktree + 提交，再 `--no-ff` 合并进主仓，冲突自动 abort； + `discard`：删 worktree 和分支。 +- `run`（resume + 新回合）、`fork`、`archive`、`list`； + `delegate`：子 thread（默认 read-only）跑完，结果可 + `--report-to-parent` 汇总回主 thread。 +- `goal`：同一 thread 多回合，直到 `--verify` 验收命令通过 + 或到 `--max-rounds`。 +- `automations --add` + `run-due`：到期任务自动派发，结果进 + host-state.json 评审队列；未到期不重复派发（幂等实测）。 +- 边界实测：read-only thread 的写文件请求被拦下、文件确实没建； + untrusted + deny 时 agent 收到审批拒绝、没拿到命令输出； + 模型内 3 子代理并行检查视角确实不同；turn/interrupt 后状态 + 为 interrupted，`compact` 需先 resume（host.py 的 compact 命令 + 已包好这一步）；合并冲突时主仓干净、冲突 worktree 留着人工解； + 隔离 CODEX_HOME 下的 MCP 工具能被发现并调用（MCP 审批是 + elicitation 协议、不是通用 decision 字段，host.py 已分流处理， + 修过一轮才通）；同一 thread 内换模型续跑、turn 的 model 字段 + 随之改变；turn/steer 能把进行中的长任务改短；3 并行 worktree + 互不污染；review 时 diff 为空也要看 untracked；不存在的模型 + 导致 task failed 并带上游 404 错误、host 不崩。 +- 工具执行修复（VM 专属）：系统裸 codex 二进制缺两个同伴 + （codex-code-mode-host 与 codex-resources/bwrap），工具会 + fail closed；配对目录 `bin/`（codex + codex-code-mode-host + + codex-resources/bwrap，约 367M，官方 0.164.0-alpha.9）经 + start.sh 优先使用；relay 支持的模型回合实测 10.3 秒内在 + work/xcm-test/test-result.txt 写出“codex-ok”。Mac 上正常 + npm 安装自带这三件套，不需要 bin/。 +- 协议层：start.sh 启动官方 codex app-server（0.164.0-alpha.9）， + 零依赖 Python stdio JSON-RPC 客户端（initialize / thread/start / + turn/start、审批、resume / fork），宿主级并行多 thread 编排， + 配置示例与官方协议 schema（generate-json-schema 导出）。 + 隔离 CODEX_HOME 下验证到模型边界为止：initialize / thread/start / + turn/start 与事件流都通，turn 随后因缺模型登录返回 401 failed， + 这是登录边界、不是协议 bug。 +- 诚实边界：云端派发、推送通知、Slack / GitHub 触发、GUI inline + diff 评论、Computer Use 都不在开源里、没有复刻；本版是本地 + host + 本地可跑的控制面演示，不是真实云服务。 + +## 诚实边界（贯穿所有版本） + +- 没有建成真实云服务：cloud/ 是本地可跑的控制面演示， + 生产仍缺常驻机器、GitHub 远端 + PAT、Slack App、云容器账号、 + 模型 key 供给方式、真实 OAuth 应用注册、真实推送凭证。 +- 没有真实 PR、没有真实推送：PR 只做到代码 + 本地可测部分 + （假 gh、本地 bare 远端）；推送只做到订阅与分发逻辑， + 缺凭证时明确报错。 +- 没有真实 token 成本：协议不暴露真实 token 数，预算按 turn + 数算；成本端点里实际与估算严格分开，估算值标 estimated。 +- 没有在任何文件、state、日志或汇报里写密钥：API key、PAT、 + Slack token、OAuth secret 一律只走环境变量或 Secure Vault， + 配置文件里只放 env_key 这类名字，不放值。 diff --git a/README.md b/README.md index 60aab1e..e3c4a68 100644 --- a/README.md +++ b/README.md @@ -198,6 +198,43 @@ turn 可以真正并发（不是轮流排队）。 或 crontab；daemon 适合常驻机器，VM 上是“会话内常驻 + 回收后重拉”的定位。 +## 宿主快速开始（最小闭环） + +下面全部是 `host.py --help` 核对过的真实子命令和参数， +不要凭印象编命令。先装检查，再走闭环；**不要在示例里 +直接派发真实模型任务来“试一下”**——dispatch 一步就会跑模型。 + +```bash +cd ~/workspace/codex-app-server +./install.sh # 只检查 python3、git、codex，不写密钥、不跑模型 +python3 host.py health # 不跑模型：查二进制、配置、state、server、daemon +python3 host.py init --repo /path/to/your/repo +# 上一步写 host-config.json（CLI > 配置 > 默认；key 绝不进配置文件，只走环境变量） +python3 host.py dispatch --name example-task --task "你要做的事" --repo /path/to/your/repo +# ^ 这一步跑模型：需要模型 key 已走环境变量提供，且会建 git worktree host/example-task +python3 host.py review --task example-task # 看 status + diff + untracked + open 评审评论 +python3 host.py merge --task example-task # 默认仓库有 tests/ 就跑 unittest，验测失败自动回滚 +# 合并不了就别硬合：python3 host.py discard --task example-task # 删 worktree 和分支 +python3 host.py daemon --interval 300 # 双 fork 常驻：每轮 recover + run-due +python3 host.py daemon-status +python3 host.py daemon-stop +``` + +三条边界，闭环前先知道： + +- `health` 不跑模型、不花额度，随时可跑；`daemon-status` / + `daemon-stop` 同样不跑模型。 +- `daemon` 在这台 VM 上活不过平台回收（和微信桥 serve 同理）： + 回收后要重拉；VM 上更稳的定时是 Muse cron 调 `health` / + `run-due`，而不是只靠 daemon 或 crontab。 +- `run-due` 到期会跑模型、会花额度：先用 + `python3 host.py automations --add --name <名> --task "<任务>" --repo <仓> --interval 86400` + 登记自动化，再确认模型 key 已走环境变量提供，才让定时去调 + `python3 host.py run-due`；未到期的自动化 run-due 时不重复派发。 +- 云端控制面演示另起一端（不跑模型回合）： + `python3 -m cloud.server`（默认 127.0.0.1:8765）， + 除 `GET /health`、`/webhooks/*`、`/oauth/*` 外都要 `X-API-Key` 请求头。 + ## GitHub PR/CI 的外部条件（真实 PR 仍缺这两样） PR/CI 这条线本轮只做了代码和本地可测部分，真实 PR 没有跑， diff --git a/VERSION b/VERSION new file mode 100644 index 0000000..6e8bf73 --- /dev/null +++ b/VERSION @@ -0,0 +1 @@ +0.1.0 diff --git a/cloud/README.md b/cloud/README.md index a018b4d..6f7a8c0 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -20,6 +20,8 @@ cloud/server.py（控制面，ThreadingHTTPServer） ├─ OAuth：cloud/oauth.py（state + 授权 URL + 回调，脚手架） ├─ 推送订阅与分发：cloud/push.py（订阅表 + 偏好过滤） ├─ 认证 / 权限 / 审计：cloud/auth.py（AuthStore） + ├─ 数据库接口：cloud/db.py（Database 接口 + SQLite 实现， + │ WAL + busy_timeout + 有限重试；Postgres 只是骨架，见下） ├─ 并发：cloud/db.py（WAL + busy_timeout + 有限重试） └─ webhook 入站：cloud/webhooks.py（验签 + 解析，只解析不执行） ``` @@ -110,6 +112,51 @@ server 的连接统一走 `cloud/db.py`：WAL 模式、busy_timeout 5000ms、 写仍是单写者串行，写密集或跨机器时必须换 Postgres，这是下一阶段， 不是本轮假装解决。 +### 数据库后端选择（接口层） + +`cloud/db.py` 已抽成接口层，不改旧调用： + +- `Database` 接口的方法就是原有 SQLite 实现对外提供的全部方法， + 按实际代码为准：`connect`（连接 + WAL 设置）、 + `execute_with_retry`（执行）、`read_all`（查询）、 + `transaction`（事务）、`is_locked_error`（锁冲突判定）。 +- `SQLiteDatabase`：现有实现改名后的类封装，原样保留 + WAL / busy_timeout / 有界重试；模块级函数 `connect` / + `execute_with_retry` / `read_all` / `transaction` / + `is_locked_error` 仍可用，行为不变。 +- `PostgresDatabase`：骨架，不是可用后端。即使驱动已装，其 + `connect` / 执行 / 查询 / 事务方法仍抛 `NotImplementedError`。 +- 工厂 `create_database(backend, db_path)` / `database_from_env()`： + 两个后端提供同一组接口方法与签名，见 + `cloud/tests/test_db_backend.py` 的接口契约测试。 + +后端选择环境变量（server 启动时生效，默认仍是 SQLite）： + +| 环境变量 | 说明 | 默认 | +| --- | --- | --- | +| `CLOUD_DB_BACKEND` | `sqlite` 或 `postgres`（`postgresql` 同义）；其它值启动即报 `ValueError`，不静默当 sqlite | `sqlite` | +| `CLOUD_DB_HOST` | Postgres 主机，全部走环境变量，不硬编码地址 | 无（未设即 None） | +| `CLOUD_DB_PORT` | Postgres 端口，须为整数 | 无（未设即 None，走驱动默认） | +| `CLOUD_DB_NAME` | Postgres 库名 | 无 | +| `CLOUD_DB_USER` | Postgres 用户名 | 无 | +| `CLOUD_DB_PASSWORD` | Postgres 密码，只走环境变量，绝不写文件、绝不进日志与 `repr` | 无 | + +选 `postgres` 且驱动缺失时启动即明确报错：`PostgresDatabase` +初始化时探测 `psycopg`（优先）/ `psycopg2`，两者都缺就抛 +`PostgresDriverMissingError`（`RuntimeError` 子类，文案含 +psycopg，写明“需安装 psycopg”），绝不静默回落到 SQLite， +也绝不假装已接通 Postgres。`create_server` 在建表之前就做这 +一步校验，所以失败时不会先建出一个 SQLite 库假装可用。 + +> 高并发换 Postgres 是下一阶段的运维项，不是本轮代码一改就 +> 自动变生产。接口层 + 骨架只解决了“以后能换”，真正上生产还 +> 需要：真实 Postgres 部署（含备份、连接数与权限）、安装 +> psycopg 驱动、写迁移脚本并把现有 SQLite 表结构与数据迁过去、 +> 把 server / AuthStore / MemoryStore / KBStore / CostStore 等 +> 目前仍直连 SQLite 的调用点逐一接到新后端并重测。在这些完成 +> 之前，即使把 `CLOUD_DB_BACKEND` 设成 `postgres`，server 也只会 +> 在启动时明确报错，不会跑起来——这是有意为之的诚实边界。 + | 方法与路径 | 权限 | 说明 | | --- | --- | --- | | `GET /health` | 无需认证 | `200 {"ok": true, "service": "cloud-control-plane"}` | @@ -225,5 +272,6 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳 端点（state + 授权 URL + 回调），但真实应用注册、回调公网地址、 已有账号绑定仍缺；推送已接成订阅 + 分发端点，但 Web Push / APNs / FCM 的真实凭证与发送端仍缺；单机 SQLite 在高并发写下 - 仍有上限，生产量上来后需换 Postgres；worker 执行、云端执行、 - 真实通知与真实模型回合均未接通。 + 仍有上限，生产量上来后需换 Postgres（接口层与骨架已就绪， + 但真实部署、驱动安装与迁移脚本仍缺，见上文“数据库后端选择”）； + worker 执行、云端执行、真实通知与真实模型回合均未接通。 diff --git a/cloud/__init__.py b/cloud/__init__.py index a4a461e..8c2e6f8 100644 --- a/cloud/__init__.py +++ b/cloud/__init__.py @@ -1,5 +1,5 @@ """云端控制面模块包。 只含 docstring，无副作用。子模块（auth、queue、sandbox、memory、kb、 -models、cost、webhooks、server）均零第三方依赖，只用 Python 标准库。 +models、cost、webhooks、server、worker）均零第三方依赖，只用 Python 标准库。 """ diff --git a/cloud/db.py b/cloud/db.py index b15ceb1..278dbf3 100644 --- a/cloud/db.py +++ b/cloud/db.py @@ -1,4 +1,14 @@ -"""SQLite 并发 helper（云端控制面单机阶段）。 +"""数据库接口层（云端控制面单机阶段）。 + +本模块分两层，调用方按需选用： + +1. 模块级函数（历史接口，原样保留，见下文）：``connect`` / + ``is_locked_error`` / ``execute_with_retry`` / ``read_all`` / + ``transaction``。server.py 与旧测试仍走这条路，行为不变。 +2. ``Database`` 接口 + 两个实现：``SQLiteDatabase``（现有 SQLite + 行为的类封装）与 ``PostgresDatabase``（骨架，见下文）。 + 新代码应走接口 + ``create_database`` 工厂，不要再直接拼 + sqlite3 连接。 只解决一件事：ThreadingHTTPServer 多线程 + 单个 SQLite 文件时的写冲突。 不碰任何业务逻辑、不定义新 schema、不改 server.py 的接线： @@ -18,10 +28,25 @@ memory.py 的 summaries 完全一致，因为本模块根本不建表）。 - WAL + busy_timeout + 有限重试只能缓解单机多线程写冲突， 不能变成多机/高并发数据库：写仍是单写者串行，读多写少尚可， 写密集或跨机器时必须换 Postgres（下一阶段，不在本轮假装解决）。 + +后端选择（环境变量，见 cloud/README.md 数据库一节）： +- ``CLOUD_DB_BACKEND``：``sqlite``（默认）或 ``postgres``。 +- Postgres 连接参数全部走环境变量，不硬编码地址或密码： + ``CLOUD_DB_HOST``、``CLOUD_DB_PORT``、``CLOUD_DB_NAME``、 + ``CLOUD_DB_USER``、``CLOUD_DB_PASSWORD``。 +- Postgres 骨架的诚实边界：标准库没有 Postgres 驱动， + ``PostgresDatabase`` 初始化时找不到 psycopg/psycopg2 会立即 + 明确报错（``PostgresDriverMissingError``，文案含 psycopg）， + 绝不静默回落到 SQLite；即使驱动已安装，其连接/执行/查询/ + 事务方法目前仍是 ``NotImplementedError`` 骨架，尚未接通真实 + Postgres——换生产数据库还需要真实部署、驱动安装与迁移脚本， + 不是改一个环境变量就自动完成。 """ +import os import sqlite3 import time +from abc import ABC, abstractmethod from contextlib import contextmanager # 与 server.py 旧 _connect 的 timeout=30s 保持一致的连接级等待； @@ -34,6 +59,18 @@ DEFAULT_MAX_RETRIES = 3 DEFAULT_BACKOFF_SECONDS = (0.05, 0.1, 0.2) _LOCKED_MARKER = "database is locked" +# Postgres 连接参数的环境变量名（名字即文档，见 cloud/README.md）。 +POSTGRES_ENV_VARS = { + "host": "CLOUD_DB_HOST", + "port": "CLOUD_DB_PORT", + "dbname": "CLOUD_DB_NAME", + "user": "CLOUD_DB_USER", + "password": "CLOUD_DB_PASSWORD", +} +BACKEND_ENV_VAR = "CLOUD_DB_BACKEND" +DEFAULT_BACKEND = "sqlite" +DEFAULT_SQLITE_PATH = "cloud-control-plane.db" + def connect(db_path, busy_timeout_ms=DEFAULT_BUSY_TIMEOUT_MS, timeout=DEFAULT_CONNECT_TIMEOUT_S): @@ -133,3 +170,220 @@ def transaction(conn): except Exception: pass raise + + +# ---------------------------------------------------------------------- +# Database 接口层 +# ---------------------------------------------------------------------- +# 方法名按实际代码为准，只覆盖模块级函数已提供的能力，不 invent +# 不存在的方法名：connect（连接 + WAL 设置）、execute_with_retry +# （执行）、read_all（查询）、transaction（事务）、is_locked_error +# （锁冲突判定）。两个后端的方法签名必须一致，见 test_db_backend.py +# 的接口契约测试（按类检查方法名与签名，不依赖驱动是否安装）。 + + +class Database(ABC): + """数据库后端接口。实现者不得静默回落到另一个后端。""" + + @abstractmethod + def connect(self): + """返回一个新连接（SQLite：已设 WAL + busy_timeout + Row）。""" + + @abstractmethod + def execute_with_retry(self, conn, sql, params=(), + max_retries=DEFAULT_MAX_RETRIES, + backoff_seconds=DEFAULT_BACKOFF_SECONDS): + """执行单条写语句，遇写锁冲突时有限重试，返回 cursor。""" + + @abstractmethod + def read_all(self, conn, sql, params=()): + """执行查询并返回全部行。""" + + @abstractmethod + def transaction(self, conn): + """返回单连接事务上下文：成功 commit，异常 rollback 后原样抛出。""" + + @abstractmethod + def is_locked_error(self, exc): + """判定异常是否为可重试的写锁冲突；非锁错误必须返回 False。""" + + +class SQLiteDatabase(Database): + """Database 的 SQLite 实现：原样保留 WAL/busy_timeout/有界重试。 + + 模块级函数仍是同一份实现，本类只是类封装，逐方法委托， + 因此旧调用与新接口看到的行为完全一致。 + """ + + def __init__(self, db_path, busy_timeout_ms=DEFAULT_BUSY_TIMEOUT_MS, + timeout=DEFAULT_CONNECT_TIMEOUT_S): + self.db_path = str(db_path) + self.busy_timeout_ms = int(busy_timeout_ms) + self.timeout = timeout + + def __repr__(self): + return ("SQLiteDatabase(db_path=%r, busy_timeout_ms=%r, " + "timeout=%r)" % (self.db_path, self.busy_timeout_ms, + self.timeout)) + + def connect(self): + return connect(self.db_path, + busy_timeout_ms=self.busy_timeout_ms, + timeout=self.timeout) + + def execute_with_retry(self, conn, sql, params=(), + max_retries=DEFAULT_MAX_RETRIES, + backoff_seconds=DEFAULT_BACKOFF_SECONDS): + return execute_with_retry(conn, sql, params, + max_retries=max_retries, + backoff_seconds=backoff_seconds) + + def read_all(self, conn, sql, params=()): + return read_all(conn, sql, params) + + @contextmanager + def transaction(self, conn): + with transaction(conn) as cur: + yield cur + + def is_locked_error(self, exc): + return is_locked_error(exc) + + +class PostgresDriverMissingError(RuntimeError): + """Postgres 驱动缺失：明确报错，绝不静默回落到 SQLite。""" + + +def _load_postgres_driver(): + """返回 (驱动名, 模块)；两者都缺时抛 PostgresDriverMissingError。 + + 优先 psycopg（v3），其次 psycopg2；只做 import 探测，不连接、 + 不读任何真实数据库、不碰任何密码。 + """ + for name in ("psycopg", "psycopg2"): + try: + module = __import__(name) + except ImportError: + continue + return name, module + raise PostgresDriverMissingError( + "Postgres 后端需安装 psycopg 驱动（例如 pip install psycopg），" + "当前环境未找到 psycopg/psycopg2；已拒绝静默回落到 SQLite，" + "请安装驱动后再选 CLOUD_DB_BACKEND=postgres。" + ) + + +_POSTGRES_NOT_IMPLEMENTED = ( + "PostgresDatabase 是骨架，尚未实现：需要真实 Postgres 部署、" + "psycopg 驱动安装与迁移脚本后才能接通，当前不假装已接通 Postgres。" +) + + +class PostgresDatabase(Database): + """Database 的 Postgres 骨架（未接通真实 Postgres）。 + + 连接参数全部走环境变量（见 POSTGRES_ENV_VARS），显式参数优先 + 于环境变量；未给出的参数保持 None，不硬编码任何地址或密码。 + 初始化时即探测驱动：缺驱动立即抛 PostgresDriverMissingError， + 因此 server 选 postgres 后端时在启动阶段就失败，不会在运行时 + 中途才炸。即使驱动已安装，connect/execute/read/transaction + 目前仍抛 NotImplementedError——骨架不是可用后端。 + 本类的 __repr__ 绝不输出 password。 + """ + + def __init__(self, host=None, port=None, dbname=None, user=None, + password=None): + env = POSTGRES_ENV_VARS + self.host = host if host is not None else os.environ.get(env["host"]) + raw_port = port if port is not None else os.environ.get(env["port"]) + if raw_port is None or raw_port == "": + self.port = None + else: + try: + self.port = int(raw_port) + except (TypeError, ValueError): + raise ValueError( + "环境变量 %s 须为整数端口，当前值无法解析" % env["port"]) + self.dbname = (dbname if dbname is not None + else os.environ.get(env["dbname"])) + self.user = (user if user is not None + else os.environ.get(env["user"])) + # 只保存、不输出、不记录日志；调用方负责传入方式的安全。 + self.password = (password if password is not None + else os.environ.get(env["password"])) + # 驱动探测在初始化时完成：缺驱动这里就报错，不等到首次查询。 + self.driver_name, self._driver = _load_postgres_driver() + + def __repr__(self): + return ("PostgresDatabase(host=%r, port=%r, dbname=%r, user=%r)" + % (self.host, self.port, self.dbname, self.user)) + + def connect(self): + raise NotImplementedError(_POSTGRES_NOT_IMPLEMENTED) + + def execute_with_retry(self, conn, sql, params=(), + max_retries=DEFAULT_MAX_RETRIES, + backoff_seconds=DEFAULT_BACKOFF_SECONDS): + raise NotImplementedError(_POSTGRES_NOT_IMPLEMENTED) + + def read_all(self, conn, sql, params=()): + raise NotImplementedError(_POSTGRES_NOT_IMPLEMENTED) + + def transaction(self, conn): + raise NotImplementedError(_POSTGRES_NOT_IMPLEMENTED) + + def is_locked_error(self, exc): + raise NotImplementedError(_POSTGRES_NOT_IMPLEMENTED) + + +def create_database(backend=None, db_path=None, **kwargs): + """后端工厂：按 backend（或环境变量）返回 Database 实现。 + + - backend 为 None 时读 ``CLOUD_DB_BACKEND``，未设则默认 sqlite。 + - ``sqlite``：返回 SQLiteDatabase（db_path 未给则用默认文件名， + 与 server 旧默认值一致）；kwargs 可传 busy_timeout_ms/timeout。 + - ``postgres``/``postgresql``：返回 PostgresDatabase；驱动缺失 + 时初始化即抛 PostgresDriverMissingError（文案含 psycopg）， + 绝不回落到 SQLite；kwargs 可传 host/port/dbname/user/password + 覆盖环境变量（测试只能用 fake/test 占位）。 + - 其它值抛 ValueError，不静默当成 sqlite。 + """ + resolved = backend + if resolved is None: + resolved = os.environ.get(BACKEND_ENV_VAR) or DEFAULT_BACKEND + resolved = str(resolved).strip().lower() + if resolved == "sqlite": + path = db_path if db_path is not None else DEFAULT_SQLITE_PATH + return SQLiteDatabase(path, **kwargs) + if resolved in ("postgres", "postgresql"): + return PostgresDatabase(**kwargs) + raise ValueError( + "未知数据库后端 %r（%s），仅支持 sqlite/postgres" + % (resolved, BACKEND_ENV_VAR)) + + +def database_from_env(db_path=None, **kwargs): + """create_database 的环境变量版：backend 一律读 CLOUD_DB_BACKEND。""" + return create_database(backend=None, db_path=db_path, **kwargs) + + +__all__ = [ + "DEFAULT_BUSY_TIMEOUT_MS", + "DEFAULT_CONNECT_TIMEOUT_S", + "DEFAULT_MAX_RETRIES", + "DEFAULT_BACKOFF_SECONDS", + "POSTGRES_ENV_VARS", + "BACKEND_ENV_VAR", + "DEFAULT_BACKEND", + "connect", + "is_locked_error", + "execute_with_retry", + "read_all", + "transaction", + "Database", + "SQLiteDatabase", + "PostgresDatabase", + "PostgresDriverMissingError", + "create_database", + "database_from_env", +] diff --git a/cloud/sandbox.py b/cloud/sandbox.py index 183eaed..9effd04 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -9,6 +9,8 @@ - CloudExecutor：占位，未接通云端执行。 """ +import json +import re import shutil import subprocess import sys @@ -50,20 +52,121 @@ class LocalExecutor(Executor): 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): + def __init__(self, host_py_path, config_path=None, runner=None, + repo=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 - def _command(self, task_id): + def _base_command(self): cmd = [sys.executable, self.host_py_path] if self.config_path is not None: cmd += ["--config", self.config_path] + 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] dispatch --name + --task [--repo ]。repo 取任务的 repo 字段， + 没有则取构造时传的 repo，都没有则不传（此时 host.py 用其 + 配置文件里的 repo，配置里也没有时 host.py 自己报错， + worker 把该报错原样写回 failed，不编造成功）。 + """ + 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() + 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] run --task + --prompt 。worker 只在 dispatch 报“任务已存在”时 + 用它续跑，不用它做首次派发。 + """ + 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() + cmd += ["run", "--task", self._host_name(task), + "--prompt", task["prompt"]] + return self._result_from_runner( + task["id"], self.runner(cmd)) + def execute(self, task) -> dict: task_id = task.get("id") if isinstance(task, dict) else None if task_id is None: diff --git a/cloud/server.py b/cloud/server.py index b1b0d86..b8645bc 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -29,6 +29,8 @@ audit_log 三张表；本 server 用 projects/tasks 两张表，表名不冲突 启动： python3 -m cloud.server [--host 127.0.0.1] [--port 8765] [--db <路径>] + 后端由 CLOUD_DB_BACKEND 选（sqlite 默认；postgres 目前只是骨架， + 驱动缺失时启动即明确报错，见 cloud/db.py）。 测试用 create_server(db_path, host, port) 只返回 ThreadingHTTPServer 实例，不自动 serve。 @@ -47,6 +49,7 @@ from urllib.parse import parse_qs, urlparse from cloud.auth import AuthStore from cloud.cost import CostStore from cloud.db import connect as _db_connect +from cloud.db import create_database as _create_database from cloud.kb import KBStore from cloud.memory import MemoryStore from cloud.oauth import ( @@ -1157,9 +1160,19 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(200, parse_slack_command(text)) -def create_server(db_path, host="127.0.0.1", port=8765): - """建 server 实例（不自动 serve，供测试与 __main__ 共用）。""" +def create_server(db_path, host="127.0.0.1", port=8765, backend=None): + """建 server 实例（不自动 serve，供测试与 __main__ 共用）。 + + backend 为 None 时读环境变量 CLOUD_DB_BACKEND，未设则默认 + sqlite（旧行为不变）。选 postgres 时在建表之前就调用工厂： + 驱动缺失则初始化即抛明确错误（文案含 psycopg），启动即失败， + 绝不在运行时中途才炸、也绝不静默回落到 SQLite；Postgres + 目前只是骨架，即使驱动已装也尚未接通，见 cloud/db.py 与 README。 + """ db_path = str(db_path) + # 后端校验必须在 init_db 之前：postgres 骨架不能先建 SQLite 表 + # 再失败，那会假装 sqlite 已可用。 + _create_database(backend=backend, db_path=db_path) init_db(db_path) # 确保 AuthStore 的三张表也存在，后续请求才能直接查 project_members。 with _auth_store(db_path): @@ -1175,8 +1188,12 @@ def main(argv=None): parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8765) parser.add_argument("--db", default="cloud-control-plane.db") + parser.add_argument("--db-backend", default=None, + help="sqlite（默认）或 postgres；未给则读 " + "CLOUD_DB_BACKEND 环境变量") args = parser.parse_args(argv) - server = create_server(args.db, args.host, args.port) + server = create_server(args.db, args.host, args.port, + backend=args.db_backend) print( "cloud-control-plane listening on %s:%d db=%s" % (args.host, server.server_address[1], args.db), diff --git a/cloud/tests/test_db_backend.py b/cloud/tests/test_db_backend.py new file mode 100644 index 0000000..0969166 --- /dev/null +++ b/cloud/tests/test_db_backend.py @@ -0,0 +1,349 @@ +"""cloud/db.py 接口层单测：Database + 双后端 + 工厂 + server 启动校验。 + +不删、不改 cloud/tests/test_db.py 的原有测试；本文件只覆盖新增的 +接口层。测试里出现的 Postgres 密码一律是 fake/test 占位 +（"test-placeholder"），不是真实格式的密钥，绝不写真实凭证。 + +覆盖点（与任务一一对应）： +- SQLite 后端原有行为不变：WAL / busy_timeout / Row / 有界重试 / + 事务提交回滚，全部经 SQLiteDatabase 实例走一遍。 +- Postgres 后端无驱动时明确报错：实例化即抛 RuntimeError 且文案含 + psycopg，且不是静默回落（返回值绝不是 SQLiteDatabase）。 +- 后端切换不丢接口契约：Database 的抽象方法集合 == 两个后端类 + 都提供的方法集合，且签名（除 self 外）一致；Postgres 方法体在 + 骨架阶段抛 NotImplementedError/驱动缺失错误。 +- server 启动校验：CLOUD_DB_BACKEND=postgres 时 create_server 在 + 建表前就报错；默认/sqlite 时仍能正常建 server。 +""" + +import inspect +import os +import sqlite3 +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +try: + from cloud.db import ( + Database, + PostgresDatabase, + PostgresDriverMissingError, + SQLiteDatabase, + create_database, + database_from_env, + ) + from cloud.server import create_server +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.db import ( + Database, + PostgresDatabase, + PostgresDriverMissingError, + SQLiteDatabase, + create_database, + database_from_env, + ) + from cloud.server import create_server + +INTERFACE_METHODS = ( + "connect", + "execute_with_retry", + "read_all", + "transaction", + "is_locked_error", +) + +TASKS_DDL = ( + "CREATE TABLE IF NOT EXISTS tasks(" + "id TEXT PRIMARY KEY, project_id TEXT, prompt TEXT, status TEXT)" +) + + +class _EnvCleanMixin: + def setUp(self): + self._env_backup = { + k: os.environ.get(k) for k in + ("CLOUD_DB_BACKEND", "CLOUD_DB_HOST", "CLOUD_DB_PORT", + "CLOUD_DB_NAME", "CLOUD_DB_USER", "CLOUD_DB_PASSWORD") + } + for key in self._env_backup: + os.environ.pop(key, None) + + def tearDown(self): + for key, value in self._env_backup.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + + +class SQLiteBackendPreservedTestCase(_EnvCleanMixin, unittest.TestCase): + def test_sqlite_database_keeps_wal_busy_timeout_row_retry(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "backend.db") + db = SQLiteDatabase(path) + conn = db.connect() + try: + self.assertEqual(conn.row_factory, sqlite3.Row) + self.assertEqual( + conn.execute("PRAGMA journal_mode").fetchone()[0], "wal") + self.assertEqual( + conn.execute("PRAGMA busy_timeout").fetchone()[0], 5000) + conn.execute(TASKS_DDL) + conn.commit() + with db.transaction(conn): + db.execute_with_retry( + conn, + "INSERT INTO tasks(id, project_id, prompt, status)" + " VALUES(?, ?, ?, 'queued')", + ("t1", "p1", "hello")) + rows = db.read_all( + conn, "SELECT * FROM tasks WHERE id = ?", ("t1",)) + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["prompt"], "hello") + # 事务回滚：重复主键失败后仍只有 1 条。 + with self.assertRaises(sqlite3.IntegrityError): + with db.transaction(conn): + db.execute_with_retry( + conn, + "INSERT INTO tasks(id, project_id, prompt," + " status) VALUES(?, ?, ?, 'queued')", + ("t1", "p1", "dup")) + self.assertEqual( + len(db.read_all(conn, "SELECT * FROM tasks")), 1) + finally: + conn.close() + # 锁判定仍走原规则：锁错误 True，其它 False。 + self.assertTrue(db.is_locked_error( + sqlite3.OperationalError("database is locked"))) + self.assertFalse(db.is_locked_error( + sqlite3.OperationalError("no such table: tasks"))) + + def test_sqlite_database_custom_busy_timeout(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "backend.db") + db = SQLiteDatabase(path, busy_timeout_ms=1234) + conn = db.connect() + try: + self.assertEqual( + conn.execute("PRAGMA busy_timeout").fetchone()[0], 1234) + finally: + conn.close() + + def test_sqlite_retry_still_bounded_via_instance(self): + # 用 fake 连接走实例方法：重试耗尽抛原始错误，不无限循环。 + class FakeConn: + def __init__(self): + self.calls = 0 + + def execute(self, sql, params=()): + self.calls += 1 + raise sqlite3.OperationalError("database is locked") + + with tempfile.TemporaryDirectory() as tmp: + db = SQLiteDatabase(os.path.join(tmp, "x.db")) + fake = FakeConn() + with mock.patch("cloud.db.time.sleep"): + with self.assertRaises(sqlite3.OperationalError): + db.execute_with_retry(fake, "INSERT ...", ()) + self.assertEqual(fake.calls, 4) # 1 次初始 + 3 次重试 + + +class PostgresSkeletonTestCase(_EnvCleanMixin, unittest.TestCase): + def test_postgres_init_without_driver_raises_psycopg_error(self): + # 本环境无 psycopg/psycopg2（已实测）；若将来装了驱动， + # 本用例应改为跳过——当前断言的是“无驱动时明确报错”。 + try: + import psycopg # noqa: F401 + has_driver = True + except ImportError: + try: + import psycopg2 # noqa: F401 + has_driver = True + except ImportError: + has_driver = False + if has_driver: + self.skipTest("环境已装 psycopg 驱动，无驱动报错路径不适用") + with self.assertRaises(PostgresDriverMissingError) as ctx: + PostgresDatabase(host="test-host", dbname="test-db", + user="test-user", + password="test-placeholder") + self.assertIn("psycopg", str(ctx.exception)) + # 绝不静默回落：异常不是 SQLite 实例，工厂同样抛错而非返回。 + with self.assertRaises(PostgresDriverMissingError) as ctx2: + create_database(backend="postgres") + self.assertIn("psycopg", str(ctx2.exception)) + self.assertIsInstance(ctx2.exception, RuntimeError) + + def test_postgres_params_come_from_env_not_hardcoded(self): + # 驱动探测 mock 成已装（不连真实 Postgres），验证参数来源： + # 环境变量进属性、显式参数优先于环境变量、repr 不泄密码。 + from cloud.db import POSTGRES_ENV_VARS + self.assertEqual(POSTGRES_ENV_VARS["host"], "CLOUD_DB_HOST") + self.assertEqual(POSTGRES_ENV_VARS["port"], "CLOUD_DB_PORT") + self.assertEqual(POSTGRES_ENV_VARS["dbname"], "CLOUD_DB_NAME") + self.assertEqual(POSTGRES_ENV_VARS["user"], "CLOUD_DB_USER") + self.assertEqual(POSTGRES_ENV_VARS["password"], "CLOUD_DB_PASSWORD") + env = { + "CLOUD_DB_HOST": "env-host", + "CLOUD_DB_PORT": "5433", + "CLOUD_DB_NAME": "env-db", + "CLOUD_DB_USER": "env-user", + "CLOUD_DB_PASSWORD": "test-placeholder", + } + with mock.patch.dict(os.environ, env): + with mock.patch("cloud.db._load_postgres_driver", + return_value=("psycopg", object())): + db = PostgresDatabase() + self.assertEqual(db.host, "env-host") + self.assertEqual(db.port, 5433) + self.assertEqual(db.dbname, "env-db") + self.assertEqual(db.user, "env-user") + self.assertNotIn("test-placeholder", repr(db)) + # 显式参数优先于环境变量；未给出的参数不硬编码成默认值。 + db2 = PostgresDatabase(host="explicit-host") + self.assertEqual(db2.host, "explicit-host") + self.assertEqual(db2.dbname, "env-db") + + def test_postgres_methods_are_skeleton_not_fake_success(self): + # 绕过 __init__（驱动探测）构造骨架实例，验证方法体本身 + # 就是明确的 NotImplementedError，而不是假装成功。 + skeleton = object.__new__(PostgresDatabase) + for name in ("connect", "read_all", "transaction", + "is_locked_error"): + method = getattr(skeleton, name) + with self.assertRaises(NotImplementedError): + if name in ("connect", "transaction"): + method(None) if name == "transaction" else method() + elif name == "is_locked_error": + method(ValueError("x")) + else: + method(None, "SELECT 1") + with self.assertRaises(NotImplementedError): + skeleton.execute_with_retry(None, "SELECT 1", ()) + + def test_postgres_repr_never_contains_password(self): + skeleton = object.__new__(PostgresDatabase) + skeleton.host = "test-host" + skeleton.port = None + skeleton.dbname = "test-db" + skeleton.user = "test-user" + skeleton.password = "test-placeholder" + self.assertNotIn("test-placeholder", repr(skeleton)) + + +class InterfaceContractTestCase(_EnvCleanMixin, unittest.TestCase): + def test_database_is_abstract(self): + with self.assertRaises(TypeError): + Database() + + def test_both_backends_expose_same_interface_methods(self): + abstract = set(Database.__abstractmethods__) + self.assertEqual(set(INTERFACE_METHODS), abstract) + for cls in (SQLiteDatabase, PostgresDatabase): + for name in INTERFACE_METHODS: + self.assertTrue( + callable(getattr(cls, name, None)), + "%s 缺少接口方法 %s" % (cls.__name__, name)) + + def test_both_backends_signatures_match_interface(self): + for cls in (SQLiteDatabase, PostgresDatabase): + for name in INTERFACE_METHODS: + expected = inspect.signature(getattr(Database, name)) + actual = inspect.signature(getattr(cls, name)) + expected_params = [p for p in expected.parameters + if p != "self"] + actual_params = [p for p in actual.parameters + if p != "self"] + self.assertEqual( + expected_params, actual_params, + "%s.%s 签名与 Database 不一致" % (cls.__name__, name)) + + def test_factory_backend_switching(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "factory.db") + # 默认（环境变量未设）仍是 SQLite。 + db = create_database(db_path=path) + self.assertIsInstance(db, SQLiteDatabase) + self.assertIsInstance(db, Database) + # 显式 sqlite。 + self.assertIsInstance( + create_database(backend="sqlite", db_path=path), + SQLiteDatabase) + # 环境变量 sqlite。 + with mock.patch.dict(os.environ, + {"CLOUD_DB_BACKEND": "sqlite"}): + self.assertIsInstance( + database_from_env(db_path=path), SQLiteDatabase) + # 未知后端：明确 ValueError，不静默当 sqlite。 + with self.assertRaises(ValueError): + create_database(backend="mysql", db_path=path) + with mock.patch.dict(os.environ, + {"CLOUD_DB_BACKEND": "mysql"}): + with self.assertRaises(ValueError): + database_from_env(db_path=path) + # 环境变量 postgres：无驱动时抛含 psycopg 的错误， + # 绝不返回一个 SQLite 实例假装切换成功。 + with mock.patch.dict(os.environ, + {"CLOUD_DB_BACKEND": "postgres"}): + try: + import psycopg # noqa: F401 + has_driver = True + except ImportError: + try: + import psycopg2 # noqa: F401 + has_driver = True + except ImportError: + has_driver = False + if has_driver: + self.skipTest("环境已装 psycopg 驱动") + with self.assertRaises(RuntimeError) as ctx: + database_from_env(db_path=path) + self.assertIn("psycopg", str(ctx.exception)) + + +class ServerBackendSelectionTestCase(_EnvCleanMixin, unittest.TestCase): + def test_server_postgres_backend_fails_at_startup(self): + try: + import psycopg # noqa: F401 + has_driver = True + except ImportError: + try: + import psycopg2 # noqa: F401 + has_driver = True + except ImportError: + has_driver = False + if has_driver: + self.skipTest("环境已装 psycopg 驱动") + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "should-not-be-created.db") + with mock.patch.dict(os.environ, + {"CLOUD_DB_BACKEND": "postgres"}): + with self.assertRaises(RuntimeError) as ctx: + create_server(path, "127.0.0.1", 0) + self.assertIn("psycopg", str(ctx.exception)) + # 启动前失败：不应已建出 SQLite 库文件假装可用。 + self.assertFalse(os.path.exists(path)) + + def test_server_default_and_sqlite_still_work(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "ok.db") + server = create_server(path, "127.0.0.1", 0) + try: + self.assertTrue(os.path.exists(path)) + finally: + server.server_close() + with mock.patch.dict(os.environ, + {"CLOUD_DB_BACKEND": "sqlite"}): + path2 = os.path.join(tmp, "ok2.db") + server2 = create_server(path2, "127.0.0.1", 0) + try: + self.assertTrue(os.path.exists(path2)) + finally: + server2.server_close() + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_worker.py b/cloud/tests/test_worker.py new file mode 100644 index 0000000..1b82e28 --- /dev/null +++ b/cloud/tests/test_worker.py @@ -0,0 +1,295 @@ +"""cloud worker 单测：认领顺序、取消、缺 key、失败写回、预算门。 + +运行（在 ~/workspace/codex-app-server 下）： + python3 -m unittest discover -s cloud/tests -p "test_worker.py" -v + +说明：只用注入的假 runner（假 CLI），不跑真实模型回合、不联网、 +不登录任何服务；env 映射也是注入的普通 dict，里面的值只是 +占位用的假名，绝不是真实密钥。 +""" + +import json +import os +import sqlite3 +import sys +import tempfile +import time +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + from cloud.cost import CostStore + from cloud.models import ModelRouter + from cloud.server import init_db + from cloud.worker import Worker, resolve_env_key +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.cost import CostStore + from cloud.models import ModelRouter + from cloud.server import init_db + from cloud.worker import Worker, resolve_env_key + + +def _insert_task(db_path, task_id, prompt="do work", status="queued", + depends_on=None, priority=0, project_id="p1"): + 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(?, ?, ?, ?, NULL, ?, ?, 'tester', ?, ?, NULL)", + (task_id, project_id, prompt, status, + json.dumps(depends_on or []), priority, now, now), + ) + conn.commit() + finally: + conn.close() + + +def _status(db_path, task_id): + conn = sqlite3.connect(db_path) + try: + row = conn.execute( + "SELECT status FROM tasks WHERE id = ?", (task_id,)).fetchone() + finally: + conn.close() + return row[0] if row else None + + +def _completed_runner(cmd): + """假 CLI：dispatch/run 一律成功，打印 host.py 风格的 JSON。""" + if "dispatch" in cmd: + name = cmd[cmd.index("--name") + 1] + else: + name = cmd[cmd.index("--task") + 1] + return SimpleNamespace( + returncode=0, + stdout=json.dumps({"task": name, "status": "completed", + "text": "done"}), + stderr="") + + +class WorkerTestCase(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": "test-placeholder"} + self.calls = [] + + def tearDown(self): + self.tmp.cleanup() + + def _worker(self, runner=None, **kwargs): + kwargs.setdefault("env", self.env) + return Worker(self.db_path, "/fake/host.py", + runner=runner or _completed_runner, **kwargs) + + def _recording_runner(self, inner): + def fake(cmd): + self.calls.append(list(cmd)) + return inner(cmd) + return fake + + # -- 认领顺序：依赖拓扑 + 优先级 ------------------------------------ + def test_claim_order_dependency_and_priority(self): + _insert_task(self.db_path, "A", priority=0) + _insert_task(self.db_path, "B", depends_on=["A"], priority=0) + _insert_task(self.db_path, "C", priority=9) + worker = self._worker(runner=self._recording_runner(_completed_runner)) + results = worker.run_all() + self.assertEqual([r["status"] for r in results], + ["completed"] * 3) + # 首波 {C, A} 按优先级降序：C 先于 A；B 依赖 A，只能最后。 + dispatch_names = [c[c.index("--name") + 1] for c in self.calls + if "dispatch" in c] + self.assertEqual(dispatch_names, ["C", "A", "B"]) + for task_id in ("A", "B", "C"): + self.assertEqual(_status(self.db_path, task_id), "completed") + + def test_dependency_on_failed_task_not_dispatched(self): + def fake(cmd): + if "dispatch" in cmd and cmd[cmd.index("--name") + 1] == "A": + return SimpleNamespace(returncode=1, stdout="boom", + stderr="") + return _completed_runner(cmd) + _insert_task(self.db_path, "A") + _insert_task(self.db_path, "B", depends_on=["A"]) + worker = self._worker(runner=fake) + results = worker.run_all() + self.assertEqual(len(results), 1) + self.assertEqual(results[0]["status"], "failed") + self.assertEqual(_status(self.db_path, "A"), "failed") + # B 的依赖未完成，始终不可派，保持 queued。 + self.assertEqual(_status(self.db_path, "B"), "queued") + + # -- 取消 ------------------------------------------------------------ + def test_cancel_queued_task_marks_interrupted(self): + _insert_task(self.db_path, "A") + worker = self._worker(runner=self._recording_runner( + _completed_runner)) + self.assertTrue(worker.cancel("A")) + self.assertEqual(_status(self.db_path, "A"), "interrupted") + # 已终态：再次取消返回 False；不存在的任务也是 False。 + self.assertFalse(worker.cancel("A")) + self.assertFalse(worker.cancel("NOPE")) + # 已取消的任务不再被派发。 + self.assertEqual(worker.run_all(), []) + self.assertEqual(self.calls, []) + + def test_cancel_during_dispatch_keeps_interrupted(self): + _insert_task(self.db_path, "A") + worker = None + + def fake(cmd): + # 执行中途从外部取消：worker 返回后不得用 completed 覆盖。 + worker.cancel("A") + return _completed_runner(cmd) + + worker = self._worker(runner=fake) + result = worker.run_one() + self.assertEqual(result["status"], "interrupted") + self.assertEqual(_status(self.db_path, "A"), "interrupted") + + # -- 缺 key 门 -------------------------------------------------------- + def test_missing_key_fails_clearly_and_does_not_run(self): + _insert_task(self.db_path, "A") + worker = self._worker( + runner=self._recording_runner(_completed_runner), env={}) + result = worker.run_one() + self.assertTrue(result["dispatched"]) + self.assertEqual(result["status"], "failed") + self.assertEqual(result["error"], + "Missing environment variable: XCM_API_KEY") + self.assertEqual(_status(self.db_path, "A"), "failed") + self.assertEqual(self.calls, []) # 缺 key 时绝不调 host.py + + def test_missing_key_empty_value_counts_as_missing(self): + _insert_task(self.db_path, "A") + worker = self._worker(env={"XCM_API_KEY": ""}) + result = worker.run_one() + self.assertEqual(result["error"], + "Missing environment variable: XCM_API_KEY") + + def test_custom_env_key_name_from_arg_and_config(self): + # 参数指定的名字出现在报错里（只含名字，不含值）。 + _insert_task(self.db_path, "A") + worker = self._worker(env={}, env_key="MY_RELAY_KEY") + self.assertEqual(worker.run_one()["error"], + "Missing environment variable: MY_RELAY_KEY") + # codex 配置文件里的 env_key 行只用来取名字。 + cfg = os.path.join(self.tmp.name, "config.toml") + with open(cfg, "w", encoding="utf-8") as f: + f.write('env_key = "CUSTOM_RELAY_KEY"\n') + self.assertEqual(resolve_env_key(codex_config_path=cfg), + "CUSTOM_RELAY_KEY") + _insert_task(self.db_path, "B") + worker2 = self._worker(env={}, codex_config_path=cfg) + self.assertEqual(worker2.env_key, "CUSTOM_RELAY_KEY") + + def test_key_present_proceeds(self): + _insert_task(self.db_path, "A") + worker = self._worker() + result = worker.run_one() + self.assertEqual(result["status"], "completed") + self.assertIsNone(result["error"]) + + # -- 执行失败写回 + 成本 ---------------------------------------------- + def test_execution_failure_writes_back_failed_and_cost(self): + def fake(cmd): + return SimpleNamespace(returncode=1, stdout="boom", stderr="") + _insert_task(self.db_path, "A", prompt="do work") + worker = self._worker(runner=fake) + result = worker.run_one() + self.assertEqual(result["status"], "failed") + self.assertIn("boom", result["error"]) + self.assertEqual(_status(self.db_path, "A"), "failed") + store = CostStore(self.db_path) + try: + report = store.report_by_task("A") + row = store._conn.execute( + "SELECT model, source, fallback_count FROM cost_records" + " WHERE task_id = ?", ("A",)).fetchone() + finally: + store._conn.close() + self.assertEqual(report["turns"], 1) + self.assertGreater(report["estimated_tokens"], 0) + self.assertEqual(report["actual_tokens"], 0) + # 复用 cost.py 的记录结构：估算标 estimated，不冒充 actual。 + self.assertEqual(row[0], "gpt-6.1-sol") + self.assertEqual(row[1], "estimated") + self.assertEqual(row[2], 0) + + # -- 预算门 ------------------------------------------------------------ + def test_budget_exceeded_no_dispatch(self): + _insert_task(self.db_path, "A") + worker = self._worker( + runner=self._recording_runner(_completed_runner), + max_turns=0) + result = worker.run_one() + self.assertFalse(result["dispatched"]) + self.assertEqual(result["reason"], "budget_exceeded") + self.assertEqual(_status(self.db_path, "A"), "queued") + self.assertEqual(self.calls, []) + + def test_cost_budget_exceeded_no_dispatch(self): + _insert_task(self.db_path, "A") + worker = self._worker(max_cost_usd=0.0) + result = worker.run_one() + self.assertFalse(result["dispatched"]) + self.assertEqual(_status(self.db_path, "A"), "queued") + + # -- fallback 事件：只记、不自动重试 ----------------------------------- + def test_timeout_fallback_event_recorded_not_retried(self): + def fake(cmd): + return SimpleNamespace( + returncode=0, + stdout=json.dumps({"task": "A", "status": "timeout", + "text": ""}), + stderr="") + _insert_task(self.db_path, "A") + router = ModelRouter(["gpt-6.1-sol", "gpt-6-sol"]) + worker = self._worker(runner=self._recording_runner(fake), + router=router) + result = worker.run_one() + self.assertEqual(result["status"], "failed") + self.assertEqual(len(result["events"]), 1) + event = result["events"][0] + self.assertEqual(event["type"], "model_fallback") + self.assertEqual(event["from_model"], "gpt-6.1-sol") + self.assertEqual(event["to_model"], "gpt-6-sol") + self.assertEqual(event["reason"], "timeout") + # 只记事件不重试：假 CLI 只被调用一次。 + self.assertEqual(len(self.calls), 1) + store = CostStore(self.db_path) + try: + report = store.report_by_task("A") + finally: + store._conn.close() + self.assertEqual(report["fallback_count"], 1) + + # -- dispatch 已存在时用 run 续跑 -------------------------------------- + def test_dispatch_existing_falls_back_to_run(self): + def fake(cmd): + if "dispatch" in cmd: + return SimpleNamespace( + returncode=1, stdout="", + stderr="Traceback: task 已存在：A，先换名或 discard 它") + return _completed_runner(cmd) + _insert_task(self.db_path, "A") + worker = self._worker(runner=self._recording_runner(fake)) + result = worker.run_one() + self.assertEqual(result["status"], "completed") + self.assertEqual(_status(self.db_path, "A"), "completed") + kinds = ["dispatch" if "dispatch" in c else "run" + for c in self.calls] + self.assertEqual(kinds, ["dispatch", "run"]) + run_cmd = [c for c in self.calls if "run" in c][0] + self.assertEqual(run_cmd[run_cmd.index("--task") + 1], "A") + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py new file mode 100644 index 0000000..585e16a --- /dev/null +++ b/cloud/worker.py @@ -0,0 +1,492 @@ +"""云端 worker：把控制面的 queued 任务接通到 host.py 执行。 + +零第三方依赖，只用 Python 标准库；不改 host.py 行为、不动 +~/.codex 配置、不碰 wechat-bridge。 + +流程（与 server.py 的任务状态机对齐）： + queued -> running -> completed / failed / interrupted + +1. 认领：pick_next() 用 queue.ready_tasks() 按依赖拓扑 + 优先级 + 排序，跳过依赖未完成（未 completed/merged）的任务；再过 + queue.budget_exceeded() 预算门，超限不派发、任务保持 queued。 + claim() 用 `WHERE status='queued'` 的 CAS 更新把任务标 running + 并记 last_heartbeat / updated_at，避免两个 worker 重复认领。 +2. 缺 key 门（执行前必查）：只检查配置的 env_key 名在进程环境 + 里是否存在且非空，绝不读出、打印、写进文件或日志 key 的值。 + 缺 key 时任务直接标 failed，错误固定为 + `Missing environment variable: `，与之前 + real-nightly-tests automation 的失败模式一致，绝不静默假装跑了。 + env_key 名的来源：显式参数 > codex 配置文件里的 env_key 行 + （只读名字）> 默认 XCM_API_KEY（中转站 relay 的约定名）。 +3. 执行：经 LocalExecutor 调 host.py 真实子命令。dispatch + （建 host 任务并跑一轮 turn）是首次派发；dispatch 报 + “任务已存在”时改用 run（在已有 host 任务上再跑一轮）续跑。 + host 任务名由 cloud 任务 id 派生（见 sandbox._host_name）， + 同一任务每次算出同一名字，续跑才能找到同一个 host 任务。 +4. 写回：server 的 tasks 表没有 result/error 列，所以执行结果 + 的持久化分三处，各复用已有结构、不另起字段名： + - tasks 表：最终状态 + updated_at（取消时为 interrupted）； + - cost.py CostStore：每轮实际执行记一条 source="estimated" + 的成本记录（host 协议不暴露真实 token 数，估算绝不冒充 + actual；total_tokens 由 CostStore.record 按 input+output 算）； + - models.py：可 fallback 的失败（如 host 状态 timeout）且 + router 给出下一个模型时，用 record_fallback() 在任务记录上 + 记 model_fallback 事件，并把 fallback_count 写进成本记录。 + 注意：worker 只记录 fallback 事件，不自动用下一个模型重试—— + 自动重试等于未经批准烧 relay 额度，重试与否由调用方决定。 + 审计：认领/完成/失败/取消/缺 key 各记一条 AuthStore 审计 + （user_id="system"，与 server.sweep_stale_tasks 的做法一致）， + 审计失败不影响任务状态（best-effort）。 + +取消：host.py 的 argparse 子命令表里没有 interrupt（turn/interrupt +只存在于 host.py Server 类内部的 app-server 协议调用，未暴露为 +CLI），所以 worker 的 cancel() 走 server 的 cancel 语义：把 +queued/running 任务直接标 interrupted（已终态的任务不动）。 +进行中的 host.py 子进程不会因此被杀（中途杀进程可能把 host +state 写一半）；run_one() 在执行返回后会先查任务是否已被取消， +已被取消的一律保持 interrupted，不用执行结果覆盖它。 + +可注入：runner（假 CLI，见 sandbox.LocalExecutor）与 env 映射 +均可注入，单测不跑真实模型回合、不联网、不登录任何服务。 +""" + +import argparse +import json +import os +import re +import time + +from cloud.auth import AuthStore +from cloud.cost import CostStore, estimate_tokens +from cloud.db import connect as _db_connect +from cloud.models import ModelRouter, is_fallbackable, record_fallback +from cloud.queue import budget_exceeded, ready_tasks +from cloud.sandbox import LocalExecutor + +# 与 server.TERMINAL_STATUSES 对齐：这些状态的任务不再被认领或取消。 +TERMINAL_STATUSES = frozenset( + {"completed", "failed", "interrupted", "merged", "merge-failed"}) +COMPLETED_STATUSES = frozenset({"completed", "merged"}) +DEFAULT_ENV_KEY = "XCM_API_KEY" +DEFAULT_MODEL = "gpt-6.1-sol" +# host.py dispatch 的原话是 "task 已存在: ，先换名或 discard 它" +#（英文 task + 已存在），取共有子串做判断，不依赖标点。 +_ALREADY_EXISTS_MARKER = "已存在" + + +def _missing_key_error(env_key): + """缺 key 的标准错误：只含 env_key 名，与 relay 侧的报错一致。""" + return "Missing environment variable: %s" % env_key + + +def resolve_env_key(env_key=None, codex_config_path=None): + """确定模型 key 对应的环境变量名（只返回名字，从不返回值）。 + + 显式参数优先；其次读 codex 配置文件（如 relay 的 config.toml） + 里的 `env_key = "..."` 行，只取名字；文件不存在或无此行时 + 用默认名。绝不读 ~/.codex/config.toml 之外的默认位置—— + 调用方要读哪个配置文件，必须显式传 codex_config_path。 + """ + if env_key: + return env_key + if codex_config_path: + try: + with open(codex_config_path, encoding="utf-8") as f: + text = f.read() + except (OSError, UnicodeDecodeError): + text = "" + match = re.search( + r'^\s*env_key\s*=\s*"([^"]+)"', text, re.MULTILINE) + if match: + return match.group(1) + return DEFAULT_ENV_KEY + + +def _resolve_model(model=None, config_path=None): + """模型名：显式参数 > host 配置文件里的 model > host 默认值。""" + if model: + return model + if config_path: + try: + with open(config_path, encoding="utf-8") as f: + cfg = json.load(f) + except (OSError, ValueError): + cfg = {} + if isinstance(cfg, dict) and cfg.get("model"): + return cfg["model"] + return DEFAULT_MODEL + + +def _task_from_row(row): + depends_on = row["depends_on"] + try: + depends_on = json.loads(depends_on) if depends_on else [] + except (TypeError, ValueError): + depends_on = [] + if not isinstance(depends_on, list): + depends_on = [depends_on] + 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"], + } + + +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): + 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 + 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() + + # -- 成本存储（惰性建表，避免只认领不执行时也建表）-------------- + @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): + conn = _db_connect(self.db_path) + try: + rows = conn.execute( + "SELECT * FROM tasks 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): + """CAS 认领：仅当任务仍为 queued 时标 running 并记心跳。""" + now = time.time() + conn = _db_connect(self.db_path) + try: + cur = conn.execute( + "UPDATE tasks SET status = 'running'," + " last_heartbeat = ?, updated_at = ?" + " WHERE id = ? AND status = 'queued'", + (now, now, task_id), + ) + 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 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 pick_next(self): + """下一个可派任务，或 None（无就绪任务 / 预算超限 / 依赖成环）。 + + 依赖满足的定义：依赖任务状态为 completed 或 merged。 + 依赖成环时 ready_tasks 抛 ValueError，worker 视为无就绪任务 + （成环任务本就永远派不出去，run_all 不应因此崩溃）。 + """ + if self._budget_exceeded(): + return None + 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 None + return ready[0] if ready else None + + # -- 取消（server cancel 语义，见模块 docstring）------------------- + def cancel(self, task_id): + """把 queued/running 任务标 interrupted；已终态或不存在返回 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) + 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 claimed: + self._audit(task["project_id"], "worker_cancelled", task_id) + return claimed + + 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 key_available(self): + """只查存在性：名在环境里且值非空即 True；值本身绝不返回。""" + try: + value = self._env.get(self.env_key) + except Exception: + return False + return value is not None and value != "" + + # -- 执行 ---------------------------------------------------------- + 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): + return {"dispatched": False, "reason": "already_claimed", + "task_id": task_id} + task = self.get_task(task_id) or task + self._audit(task["project_id"], "worker_claim", task_id) + + # 缺 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._audit(task["project_id"], "worker_missing_key", error) + return {"dispatched": True, "task_id": task_id, + "host_name": self.executor._host_name(task) + if hasattr(self.executor, "_host_name") else task_id, + "status": "failed", "executor": "local", + "stdout": "", "error": error, "events": [], + "cost_record_id": None} + + events = [] + fallback_count = 0 + result = None + error = None + try: + 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: + # host 任务已存在（上轮 worker 在 dispatch 后、写回前 + # 崩溃）：改用 run 在已有任务上续跑，不重复建 worktree。 + 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) + + # 取消优先：执行期间被 cancel 的任务保持 interrupted， + # 绝不用执行结果覆盖它（见模块 docstring 的取消说明）。 + if self._was_cancelled(task_id): + self._set_status(task_id, "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": "local", + "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 error is None: + error = stdout.strip() or "host execution failed" + # fallback 事件：只记、不自动重试（见模块 docstring）。 + if self.router is not None and host_status is not None \ + and is_fallbackable(host_status): + nxt = self.router.next_after(self.model, host_status) + if nxt is not None: + record = dict(task) + record["events"] = events + # record_fallback 本身会 append 到 events， + # 这里不要再 append 一次，否则事件会重复。 + record_fallback( + record, self.model, nxt, str(host_status)) + fallback_count = 1 + self._audit(task["project_id"], "worker_fallback", + "%s -> %s" % (self.model, nxt)) + self._set_status(task_id, final) + self.usage_turns += 1 + # 成本：host 协议不暴露真实 token 数，用 cost.py 的估算结构 + # 记 estimated，绝不标 actual；fallback_count 复用同一记录。 + total = estimate_tokens(task.get("prompt") or "", stdout) + cost_id = self.costs.record( + task_id, task["project_id"], self.model, + input_tokens=total, output_tokens=0, usd_micros=None, + source="estimated", fallback_count=fallback_count) + self._audit(task["project_id"], + "worker_completed" if final == "completed" + else "worker_failed", 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": "local", + "stdout": stdout, "error": error, "events": events, + "cost_record_id": cost_id} + + def run_all(self, max_tasks=None): + """循环认领执行，直到无就绪任务、预算超限或达到 max_tasks。 + + 每轮都重新读库，所以上一轮刚 completed 的任务会解锁它的 + 依赖任务；每轮至少推进一个任务到终态，不会空转死循环。 + """ + 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 + + +def main(argv=None): + """单次派发循环：python3 -m cloud.worker --db <库> --host-py <路径>。 + + 只打印任务结果摘要（状态、host 名、错误文本），绝不打印 + 任何环境变量的值。 + """ + parser = argparse.ArgumentParser(description="cloud worker (one pass)") + 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("--env-key", default=None) + parser.add_argument("--codex-config", default=None, + help="含 env_key 行的 codex 配置文件（只读名字）") + parser.add_argument("--model", default=None) + parser.add_argument("--max-turns", type=int, default=None) + parser.add_argument("--max-tasks", type=int, default=None) + args = parser.parse_args(argv) + 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, + max_turns=args.max_turns) + try: + 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() diff --git a/install.sh b/install.sh new file mode 100755 index 0000000..3e447e4 --- /dev/null +++ b/install.sh @@ -0,0 +1,70 @@ +#!/bin/bash +# codex-app-server 安装检查（不装包、不写密钥、不跑模型）。 +# +# 用法： +# ./install.sh +# +# 通过后怎么启动（见 README“宿主快速开始”）： +# python3 host.py health # 不跑模型，只做健康检查 +# python3 host.py init --repo <你的仓库路径> +# python3 -m cloud.server # 云端控制面演示，默认 127.0.0.1:8765 +# +# 三条硬规则（脚本本身也遵守）： +# 1. bin/codex 约 367M 不进仓库：.gitignore 已排除 bin/，安装时需 +# 自行配对官方二进制（本机 VM 做法见下；Mac 上 npm 正常安装 +# 自带 codex-code-mode-host 与 codex-resources/bwrap，不需要 bin/）。 +# 2. 模型 key 只走环境变量：配置文件里只放 env_key 这类名字 +# （例如 XCM_API_KEY 这类名字），绝不在脚本、配置、日志里写 key 的值。 +# 3. 本脚本不登录任何账号、不向任何外部服务发请求、不跑模型回合。 +set -euo pipefail +cd "$(dirname "$0")" + +fail=0 + +# 1. python3 是否存在 +if command -v python3 >/dev/null 2>&1; then + echo "ok: python3 $(python3 --version 2>&1)" +else + echo "missing: python3（请先安装 Python 3）" >&2 + fail=1 +fi + +# 2. git 是否存在（dispatch / review / merge 走 git worktree） +if command -v git >/dev/null 2>&1; then + echo "ok: git $(git --version 2>&1)" +else + echo "missing: git（host.py 的 worktree 隔离与合并门需要 git）" >&2 + fail=1 +fi + +# 3. 官方 codex 二进制是否可用（只检查存在性，不启动、不登录） +if [ -n "${CODEX_BIN:-}" ] && [ -x "$CODEX_BIN" ]; then + echo "ok: codex via CODEX_BIN=$CODEX_BIN" +elif [ -x "bin/codex" ]; then + echo "ok: codex via bin/codex（本地配对三件套，约 367M，不进仓库）" +elif command -v codex >/dev/null 2>&1; then + echo "ok: codex via PATH: $(command -v codex)" +else + echo "warn: 未找到 codex（二进制需自行配对官方安装；" + echo " 本机 VM 曾用 bin/ 配对 codex + codex-code-mode-host +" + echo " codex-resources/bwrap 三件套，Mac 上 npm 安装自带三件套）" + echo " 没有 codex 时 host.py health 会报失败，dispatch 会跑不起来，" + echo " 但本脚本不替你下载或登录。" +fi + +# 4. 提醒：模型 key 只走环境变量，这里只检查名字是否被设置过方向， +# 绝不打印、绝不写入任何 key 的值。 +if [ -n "${XCM_API_KEY:-}" ]; then + echo "ok: 模型 key 已通过环境变量提供（名字：XCM_API_KEY；值不在此打印）" +else + echo "note: 未检测到 XCM_API_KEY 这类环境变量名——跑模型回合" + echo " （dispatch / run-due 到期派发）前需先在你自己的 shell 里" + echo " export 相应变量；health 不跑模型，不需要 key。" +fi + +if [ "$fail" -ne 0 ]; then + echo "install check FAILED：补齐上面 missing 的项后再跑。" >&2 + exit 1 +fi + +echo "install check OK：接下来跑 python3 host.py health，再按 README 快速开始走 init → dispatch → review → merge。"