feat: per-task model/effort/approval + relay models endpoint (#59)

Co-authored-by: host <host@test.local>
This commit is contained in:
openqjandhost authored and GitHub committed 2026-10-27 01:44:53 +08:00
1 parent cff152c5a2
commit 8b2f3968dc
9 files changed
+521 -16

No files matched your search

+2 -1
View File
@@ -366,8 +366,9 @@ psycopg,写明“需安装 psycopg”),绝不静默回落到 SQLite,
| `GET /health` | 无需认证 | `200 {"ok": true, "service": "cloud-control-plane"}` |
| `POST /projects` | 认证即可 | body `{name, id?}`;无 id 则生成 uuid;调用者记为 owner(调 `AuthStore.create_project`,同时写 projects 表并记审计 `create_project`);返回 `201` + 项目 |
| `GET /projects` | 认证即可 | 只返回调用者是成员的项目:`{"projects": [...]}` |
| `POST /tasks` | `create_task` | body `{project_id, prompt, depends_on?, priority?, idempotency_key?, executor?, goal?, verify_command?, goal_max_rounds?}`;无权限 `403`;同 `project_id + idempotency_key` 已存在则返回已有任务(`200`,不建重复);否则建 `queued` 任务并记审计 `create_task`,返回 `201`;`verify_command` 非空即 goal 模式(见下文“goal 模式”),`goal_max_rounds` 缺省 3、须 ≥1 |
| `POST /tasks` | `create_task` | body `{project_id, prompt, depends_on?, priority?, idempotency_key?, executor?, goal?, verify_command?, goal_max_rounds?, model?, effort?, approval?}`;无权限 `403`;同 `project_id + idempotency_key` 已存在则返回已有任务(`200`,不建重复);否则建 `queued` 任务并记审计 `create_task`,返回 `201`;`verify_command` 非空即 goal 模式(见下文“goal 模式”),`goal_max_rounds` 缺省 3、须 ≥1;`model` 不做白名单(中转目录会变,console 下拉只是建议值),`effort` 只能是 `low/medium/high/xhigh/max`,`approval` 只能是 `ask/auto/full`,三者缺省均为 NULL(= 跟随 worker 配置默认值,行为与过去一致) |
| `GET /tasks?project_id=...` | 该项目 `read` | 返回该项目任务列表:`{"tasks": [...]}`;无权限 `403` |
| `GET /relay/models` | 任意已认证用户 | 中转站可用模型列表:服务端凭环境变量 `XCM_API_KEY` 现取 `XCM_RELAY_BASE_URL`(缺省 `https://xcmapi.org/v1`)的 `/v1/models`,只返回模型 id(1 小时缓存),key 绝不进响应;中转挂了或未配 key 时 `502`,不编造列表;console 的模型下拉用它,取不到时回退手动输入 |
| `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` |
+49
View File
@@ -86,6 +86,26 @@
<label><input id="goalMode" type="checkbox"> goal 模式</label>
</div>
<textarea id="prompt" placeholder="任务 prompt(必填)"></textarea>
<div class="bar">
<select id="modelSelect">
<option value="">模型:默认(跟随配置)</option>
</select>
<input id="modelCustom" type="text" placeholder="手动输入模型(中转列表取不到时用)" style="display:none">
<select id="effortSelect">
<option value="">推理强度:默认(跟随配置)</option>
<option value="low">low</option>
<option value="medium">medium</option>
<option value="high">high</option>
<option value="xhigh">xhigh</option>
<option value="max">max</option>
</select>
<select id="approvalSelect">
<option value="">权限:默认(跟随配置)</option>
<option value="ask">请求批准</option>
<option value="auto">帮我批准</option>
<option value="full">完全访问</option>
</select>
</div>
<div id="goalFields" style="display:none">
<input id="goal" type="text" placeholder="goal(缺省用 prompt)" style="width:100%; box-sizing:border-box; margin:6px 0;">
<input id="verifyCommand" type="text" placeholder="verify_command 验收命令(goal 模式必填项按服务端规则)" style="width:100%; box-sizing:border-box; margin:6px 0;">
@@ -157,6 +177,23 @@ async function api(path, method, body) {
return data;
}
async function loadRelayModels() {
// 中转模型下拉:服务端现取现回,只含模型 id。取不到时不挡
// 提交,留手动输入框给用户填。
var sel = $("#modelSelect");
try {
var d = await api("/relay/models", "GET");
var models = d.models || [];
sel.innerHTML = "<option value=''>模型:默认(跟随配置)</option>" +
models.map(function (m) {
return "<option value='" + esc(m) + "'>" + esc(m) + "</option>";
}).join("");
$("#modelCustom").style.display = "none";
} catch (e) {
$("#modelCustom").style.display = "inline";
}
}
async function loadHealth() {
try {
var h = await api("/health", "GET");
@@ -229,6 +266,8 @@ async function showDetail(taskId) {
$("#detail").innerHTML =
"状态 <b class='st-" + esc(t.status) + "'>" + esc(t.status) + "</b> | " +
"executor " + esc(t.executor || "") + " | 主机 " + esc(t.worktree_host || "") +
" | 模型 " + esc(t.model || "默认") + " | 强度 " + esc(t.effort || "默认") +
" | 权限 " + esc(t.approval || "默认") +
(t.pr_url ? " | <a href='" + esc(t.pr_url) + "' target='_blank' rel='noopener'>PR #" + esc(t.pr_number) + "</a>" : "") +
"<br>prompt:<br>" + esc(t.prompt || "");
var body = "";
@@ -268,6 +307,7 @@ $("#saveKey").onclick = function () {
loadProjects();
loadWorkers();
if ($("#projectSelect").value) loadTasks();
if (getKey()) loadRelayModels();
};
$("#clearKey").onclick = function () { localStorage.removeItem(KEY_STORE); setKeyState(); };
$("#refreshHealth").onclick = loadHealth;
@@ -297,6 +337,15 @@ $("#createTask").onclick = async function () {
var body = { project_id: pid, prompt: prompt };
var ex = $("#executor").value;
if (ex) body.executor = ex;
var model = $("#modelCustom").value.trim() || $("#modelSelect").value;
if (model) body.model = model;
var eff = $("#effortSelect").value;
if (eff) body.effort = eff;
var ap = $("#approvalSelect").value;
if (ap) {
if (ap === "full" && !confirm("完全访问 = 自动批准 + 不受工作区限制,agent 可读写工作区外路径。确认提交吗?")) return;
body.approval = ap;
}
if ($("#goalMode").checked) {
body.goal = $("#goal").value.trim() || prompt;
body.verify_command = $("#verifyCommand").value.trim();
+3
View File
@@ -831,6 +831,9 @@ POSTGRES_SCHEMA = (
"goal TEXT, "
"verify_command TEXT, "
"goal_max_rounds INTEGER, "
"model TEXT, "
"effort TEXT, "
"approval TEXT, "
"UNIQUE(project_id, idempotency_key))",
"CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status)",
"CREATE TABLE IF NOT EXISTS summaries("
+77
View File
@@ -0,0 +1,77 @@
"""从中转站(OpenAI 兼容 /v1/models)取可用模型列表。
只返回模型 id,不返回、不记录任何密钥:key 只进请求头,
绝不进返回值、日志或缓存。列表缓存 1 小时(中转模型目录
很少变),中转挂了时调用方拿到缓存;无缓存又取不到时报
RelayError,由端点翻成 502。
base_url:环境变量 XCM_RELAY_BASE_URL,缺省
https://xcmapi.org/v1(与 codex-home 里 relay 配置的
base_url 一致)。key:环境变量 XCM_API_KEY(与
config.xcmapi.example.toml 的 env_key 一致)。
"""
import json
import os
import time
import urllib.error
import urllib.request
DEFAULT_BASE_URL = "https://xcmapi.org/v1"
CACHE_TTL_SECONDS = 3600
class RelayError(RuntimeError):
pass
_cache = {"models": None, "updated_at": 0.0}
def _default_fetcher(url, headers, timeout):
req = urllib.request.Request(url, headers=headers, method="GET")
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
return json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as exc:
raise RelayError("relay models HTTP %s" % exc.code)
except (urllib.error.URLError, ValueError, TimeoutError) as exc:
raise RelayError("relay models fetch failed: %s" % exc)
def get_relay_models(fetcher=None, now=None):
"""返回 (models, cached, updated_at):models 为 id 字符串列表。
fetcher 可注入(单测用):签名 fetcher(url, headers, timeout)
-> 已解析的 JSON dict。缓存命中时不调 fetcher。
"""
now = time.time() if now is None else now
if (_cache["models"] is not None
and now - _cache["updated_at"] < CACHE_TTL_SECONDS):
return list(_cache["models"]), True, _cache["updated_at"]
base_url = os.environ.get("XCM_RELAY_BASE_URL") or DEFAULT_BASE_URL
url = base_url.rstrip("/") + "/models"
if fetcher is not None:
# 单测注入:不带 key、不碰环境,生产路径仍走下面强制 key。
data = fetcher(url, {}, 10)
else:
api_key = os.environ.get("XCM_API_KEY")
if not api_key:
raise RelayError("relay key not configured")
data = _default_fetcher(
url, {"Authorization": "Bearer " + api_key}, 10)
if not isinstance(data, dict) or not isinstance(data.get("data"), list):
raise RelayError("relay models unexpected response")
models = []
for item in data["data"]:
model_id = item.get("id") if isinstance(item, dict) else None
if isinstance(model_id, str) and model_id and model_id not in models:
models.append(model_id)
_cache["models"] = list(models)
_cache["updated_at"] = now
return list(models), False, now
def _reset_cache_for_tests():
_cache["models"] = None
_cache["updated_at"] = 0.0
+51 -11
View File
@@ -133,6 +133,27 @@ def _default_runner(cmd):
return subprocess.run(cmd, capture_output=True, text=True)
def _approval_cli_args(approval):
"""console 权限三档 → host.py 顶层全局参数。
请求批准(ask):approvalPolicy=untrusted,敏感操作先问;
帮我批准(auto):approvalPolicy=never,自动批准,仍受
工作区沙箱限制;完全访问(full):approvalPolicy=never
+ sandbox=danger-full-access,不再受工作区限制(危险)。
None 或未知值:不传参数,host.py 走配置文件默认值。
docker 任务的配置沙箱本身就是 danger-full-access,所以
full 档对 docker 任务无变化,变化只在 approvalPolicy。
"""
if approval == "ask":
return ["--approval-policy", "untrusted"]
if approval == "auto":
return ["--approval-policy", "never"]
if approval == "full":
return ["--approval-policy", "never",
"--sandbox", "danger-full-access"]
return []
class LocalExecutor(Executor):
"""本地执行器:调 host.py CLI 子进程做执行封装。
@@ -166,13 +187,16 @@ class LocalExecutor(Executor):
# 不改 host.py 的 dispatch/run 签名。
self.model = model
def _base_command(self, model=None):
def _base_command(self, model=None, effort=None, approval=None):
cmd = [sys.executable, self.host_py_path]
if self.config_path is not None:
cmd += ["--config", self.config_path]
effective = model or self.model
if effective:
cmd += ["--model", effective]
if effort:
cmd += ["--effort", effort]
cmd += _approval_cli_args(approval)
return cmd
def _command(self, task_id):
@@ -256,7 +280,9 @@ class LocalExecutor(Executor):
raise ValueError("任务缺少 id")
if not task.get("prompt"):
raise ValueError("任务缺少 prompt")
cmd = self._base_command(model=task.get("model"))
cmd = self._base_command(model=task.get("model"),
effort=task.get("effort"),
approval=task.get("approval"))
cmd += ["dispatch", "--name", self._host_name(task),
"--task", task["prompt"]]
repo = task.get("repo") or self.repo
@@ -277,7 +303,9 @@ class LocalExecutor(Executor):
raise ValueError("任务缺少 id")
if not task.get("prompt"):
raise ValueError("任务缺少 prompt")
cmd = self._base_command(model=task.get("model"))
cmd = self._base_command(model=task.get("model"),
effort=task.get("effort"),
approval=task.get("approval"))
cmd += ["run", "--task", self._host_name(task),
"--prompt", task["prompt"]]
return self._result_from_runner(
@@ -303,7 +331,9 @@ class LocalExecutor(Executor):
if not goal_text:
raise ValueError("任务缺少 prompt")
rounds = task.get("goal_max_rounds") or 3
cmd = self._base_command(model=task.get("model"))
cmd = self._base_command(model=task.get("model"),
effort=task.get("effort"),
approval=task.get("approval"))
cmd += ["goal", "--name", self._host_name(task),
"--task", goal_text, "--verify", verify,
"--max-rounds", str(rounds)]
@@ -371,7 +401,9 @@ class LocalExecutor(Executor):
raise ValueError("任务缺少 id")
if not task.get("prompt"):
raise ValueError("任务缺少 prompt")
cmd = self._base_command(model=task.get("model"))
cmd = self._base_command(model=task.get("model"),
effort=task.get("effort"),
approval=task.get("approval"))
if kind == "run":
cmd += ["run", "--task", self._host_name(task),
"--prompt", task["prompt"]]
@@ -636,7 +668,8 @@ class DockerExecutor(Executor):
cmd += list(inner_cmd)
return cmd
def _inner_host_command(self, *args, model=None):
def _inner_host_command(self, *args, model=None, effort=None,
approval=None):
"""容器内 host.py 命令:python3 /app/host.py [args...]。
config_path 若已给,容器内用 /app/<basename> 指代它——
@@ -644,8 +677,9 @@ class DockerExecutor(Executor):
本就放在项目目录);不在代码目录时不静默猜路径,
由 _inner_host_command 的调用方在传 config 前自行确认。
这里只做最简单的同名映射,不展开宿主绝对路径进容器。
--model 是 host.py 的顶层全局参数,必须放在子命令之前,
所以 model 在这里插到 --config 之后、子命令之前。
--model/--effort/--approval-policy/--sandbox 是 host.py
的顶层全局参数,必须放在子命令之前,所以在这里插到
--config 之后、子命令之前。
"""
cmd = ["python3", "/app/host.py"]
if self.config_path is not None:
@@ -654,6 +688,9 @@ class DockerExecutor(Executor):
effective = model or self.model
if effective:
cmd += ["--model", effective]
if effort:
cmd += ["--effort", effort]
cmd += _approval_cli_args(approval)
cmd += list(args)
return cmd
@@ -790,7 +827,8 @@ class DockerExecutor(Executor):
raise ValueError("任务缺少 prompt")
inner = self._inner_host_command(
"dispatch", "--name", self._host_name(task),
"--task", task["prompt"], model=task.get("model"))
"--task", task["prompt"], model=task.get("model"),
effort=task.get("effort"), approval=task.get("approval"))
repo = task.get("repo")
if repo:
inner += ["--repo", repo]
@@ -808,7 +846,8 @@ class DockerExecutor(Executor):
raise ValueError("任务缺少 prompt")
inner = self._inner_host_command(
"run", "--task", self._host_name(task),
"--prompt", task["prompt"], model=task.get("model"))
"--prompt", task["prompt"], model=task.get("model"),
effort=task.get("effort"), approval=task.get("approval"))
return self._run_in_container(task, inner, timeout=timeout,
extra_mounts=extra_mounts,
network=network,
@@ -836,7 +875,8 @@ class DockerExecutor(Executor):
inner = self._inner_host_command(
"goal", "--name", self._host_name(task),
"--task", goal_text, "--verify", verify,
"--max-rounds", str(rounds), model=task.get("model"))
"--max-rounds", str(rounds), model=task.get("model"),
effort=task.get("effort"), approval=task.get("approval"))
repo = task.get("repo")
if repo:
inner += ["--repo", repo]
+54 -3
View File
@@ -83,6 +83,8 @@ from cloud.oauth import (
start_oauth_login,
unbind_external_identity,
)
from cloud.relay import RelayError as _RelayError
from cloud.relay import get_relay_models as _get_relay_models
from cloud.repos import validate_repo_url
from cloud.push import (
create_subscription as _push_create,
@@ -389,6 +391,12 @@ def migrate_tasks_table(conn):
# HA 阶段 1:认领任务的 worker 所在主机名;/pr 据此判断
# worktree 在本机(同步跑)还是别机(排 pr 任务给该主机)。
"worktree_host": "TEXT",
# 任务级选项(console 选模型/强度/权限用):NULL = 跟随
# worker 配置默认值;model 不做白名单(中转目录会变,
# 下拉只是建议值),effort/approval 由 _create_task 校验。
"model": "TEXT",
"effort": "TEXT",
"approval": "TEXT",
}
for name, coltype in new_cols.items():
if name not in cols:
@@ -524,6 +532,9 @@ def _task_to_dict(row):
"verify_command": _row_get(row, "verify_command"),
"goal_max_rounds": _row_get(row, "goal_max_rounds"),
"worktree_host": _row_get(row, "worktree_host"),
"model": _row_get(row, "model"),
"effort": _row_get(row, "effort"),
"approval": _row_get(row, "approval"),
}
@@ -817,6 +828,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
self._metrics(user_id)
elif path == "/models" and method == "GET":
self._models_overview(user_id)
elif path == "/relay/models" and method == "GET":
self._relay_models(user_id)
elif path.startswith("/projects/"):
self._dispatch_project(user_id, method, path, query)
elif path == "/tasks" and method == "POST":
@@ -1295,6 +1308,20 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
self._send_json(200, {"task_id": task_id,
"project_id": task["project_id"], **report})
def _relay_models(self, user_id):
# 中转可用模型列表:服务端凭环境变量里的 relay key 现取
# 现回,只把模型 id 交给调用者。key 绝不进响应、日志、
# 缓存(见 cloud/relay.py)。单测可经 server.relay_fetcher
# 注入假取数函数;中转挂了或未配 key 时 502,不编造列表。
fetcher = getattr(self.server, "relay_fetcher", None)
try:
models, cached, updated_at = _get_relay_models(fetcher=fetcher)
except _RelayError as exc:
self._send_json(502, {"error": str(exc)})
return
self._send_json(200, {
"models": models, "cached": cached, "updated_at": updated_at})
def _models_overview(self, user_id):
# 优先级列表来自环境变量 MODELS_PRIORITY(逗号分隔),未设时
# 用默认列表;这只是控制面的默认配置,不是 provider 的承诺。
@@ -1382,6 +1409,28 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
self._send_json(400, {
"error": "executor must be local, docker or cloud"})
return
# 任务级选项:model 不做白名单(中转目录会变,下拉只是
# 建议值);effort 必须是 codex 的 effort 档位;approval
# 是 console 三档(ask/auto/full),非法值 400 且不建任务。
# 三者都缺省(NULL)时 worker 走配置默认值,行为与过去一致。
model = body.get("model")
if model is not None:
if not isinstance(model, str) or not model.strip():
self._send_json(400, {
"error": "model must be a non-empty string"})
return
model = model.strip()
effort = body.get("effort")
if effort is not None and effort not in (
"low", "medium", "high", "xhigh", "max"):
self._send_json(400, {
"error": "effort must be low, medium, high, xhigh or max"})
return
approval = body.get("approval")
if approval is not None and approval not in ("ask", "auto", "full"):
self._send_json(400, {
"error": "approval must be ask, auto or full"})
return
# goal 字段:verify_command 非空即 goal 模式(worker 走
# host.py goal 循环);goal 文本缺省用 prompt;轮次缺省 3
# (与 host.py 默认一致),非法值 400 且不建任务。
@@ -1431,12 +1480,14 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
"INSERT INTO tasks(id, project_id, prompt, status,"
" idempotency_key, depends_on, priority, created_by,"
" created_at, updated_at, last_heartbeat, executor,"
" goal, verify_command, goal_max_rounds)"
" goal, verify_command, goal_max_rounds,"
" model, effort, approval)"
" VALUES(?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, NULL, ?,"
" ?, ?, ?)",
" ?, ?, ?, ?, ?, ?)",
(task_id, project_id, prompt, idempotency_key,
json.dumps(depends_on), priority, user_id, now, now,
executor, goal, verify_command, goal_max_rounds),
executor, goal, verify_command, goal_max_rounds,
model, effort, approval),
)
conn.commit()
except (sqlite3.IntegrityError, _PgIntegrityError):
+81
View File
@@ -0,0 +1,81 @@
"""中转模型列表(cloud/relay.py)单测:只用注入的取数函数,不联网、
不碰真 key。"""
import os
import sys
import unittest
from pathlib import Path
from unittest import mock
try:
from cloud.relay import RelayError, get_relay_models
from cloud.relay import _reset_cache_for_tests
except ImportError:
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.relay import RelayError, get_relay_models
from cloud.relay import _reset_cache_for_tests
class RelayTestCase(unittest.TestCase):
def setUp(self):
_reset_cache_for_tests()
self.env = mock.patch.dict(os.environ, {}, clear=False)
self.env.start()
os.environ.pop("XCM_API_KEY", None)
def tearDown(self):
self.env.stop()
_reset_cache_for_tests()
def test_parses_ids_and_dedups(self):
calls = []
def fetcher(url, headers, timeout):
calls.append((url, headers))
return {"data": [{"id": "gpt-6.1-sol"}, {"id": "gpt-6-sol"},
{"id": "gpt-6.1-sol"}, {"nope": 1}]}
models, cached, _ = get_relay_models(fetcher=fetcher)
self.assertEqual(models, ["gpt-6.1-sol", "gpt-6-sol"])
self.assertFalse(cached)
self.assertTrue(calls[0][0].endswith("/models"))
# 注入路径不带 key:响应与请求头里都不应出现密钥字样。
self.assertNotIn("Authorization", calls[0][1])
self.assertEqual(len(calls), 1)
def test_cache_hit_skips_second_fetch(self):
calls = []
def fetcher(url, headers, timeout):
calls.append(url)
return {"data": [{"id": "m1"}]}
models1, cached1, _ = get_relay_models(fetcher=fetcher)
models2, cached2, _ = get_relay_models(fetcher=fetcher)
self.assertEqual(models1, models2)
self.assertFalse(cached1)
self.assertTrue(cached2)
self.assertEqual(len(calls), 1)
def test_cache_expires_after_ttl(self):
calls = []
def fetcher(url, headers, timeout):
calls.append(url)
return {"data": [{"id": "m1"}]}
get_relay_models(fetcher=fetcher, now=1000.0)
_, cached, _ = get_relay_models(fetcher=fetcher, now=1000.0 + 3601)
self.assertFalse(cached)
self.assertEqual(len(calls), 2)
def test_missing_key_without_fetcher_raises(self):
with self.assertRaises(RelayError):
get_relay_models()
def test_bad_response_raises(self):
with self.assertRaises(RelayError):
get_relay_models(fetcher=lambda u, h, t: {"nope": []})
with self.assertRaises(RelayError):
get_relay_models(fetcher=lambda u, h, t: {"data": "x"})
if __name__ == "__main__":
unittest.main()
+196
View File
@@ -0,0 +1,196 @@
"""任务级选项(model/effort/approval)端到端单测。
覆盖:POST /tasks 校验与落库回读、GET /relay/models(注入取数,
不联网不碰真 key)、LocalExecutor/DockerExecutor 命令行里
的 --model/--effort/--approval-policy/--sandbox 位置与映射。
不跑真实模型,不启动 worker。
"""
import http.client
import json
import os
import sys
import tempfile
import threading
import unittest
from pathlib import Path
from types import SimpleNamespace
try:
from cloud.auth import AuthStore
from cloud.relay import _reset_cache_for_tests
from cloud.sandbox import DockerExecutor, LocalExecutor
from cloud.sandbox import _approval_cli_args
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.relay import _reset_cache_for_tests
from cloud.sandbox import DockerExecutor, LocalExecutor
from cloud.sandbox import _approval_cli_args
from cloud.server import create_server
class TaskOptionsTestCase(unittest.TestCase):
def setUp(self):
_reset_cache_for_tests()
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.server = create_server(self.db_path, "127.0.0.1", 0)
self.server.relay_fetcher = lambda u, h, t: {
"data": [{"id": "gpt-6.1-sol"}, {"id": "gpt-6-astra"}]}
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)
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()
_reset_cache_for_tests()
def _req(self, method, path, body=None, key=None):
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
headers = {"Content-Type": "application/json"}
if key:
headers["X-API-Key"] = key
conn.request(method, path,
json.dumps(body) if body is not None else None, headers)
resp = conn.getresponse()
raw = resp.read()
conn.close()
try:
data = json.loads(raw)
except ValueError:
data = None
return resp.status, data
def test_create_task_with_options_roundtrip(self):
status, task = self._req("POST", "/tasks", {
"project_id": self.project["id"], "prompt": "hi",
"model": "gpt-6-astra", "effort": "high",
"approval": "full"}, self.owner_key)
self.assertEqual(status, 201)
self.assertEqual(task["model"], "gpt-6-astra")
self.assertEqual(task["effort"], "high")
self.assertEqual(task["approval"], "full")
status, tasks = self._req(
"GET", "/tasks?project_id=" + self.project["id"],
key=self.owner_key)
self.assertEqual(status, 200)
self.assertEqual(tasks["tasks"][0]["model"], "gpt-6-astra")
def test_create_task_defaults_are_null(self):
status, task = self._req("POST", "/tasks", {
"project_id": self.project["id"], "prompt": "hi"},
self.owner_key)
self.assertEqual(status, 201)
self.assertIsNone(task["model"])
self.assertIsNone(task["effort"])
self.assertIsNone(task["approval"])
def test_create_task_invalid_options_rejected(self):
for body in ({"effort": "ultra"}, {"approval": "yes"},
{"model": ""}, {"model": 42}):
payload = {"project_id": self.project["id"], "prompt": "hi"}
payload.update(body)
status, data = self._req("POST", "/tasks", payload,
self.owner_key)
self.assertEqual(status, 400, body)
def test_relay_models_endpoint(self):
status, data = self._req("GET", "/relay/models", key=self.owner_key)
self.assertEqual(status, 200)
self.assertEqual(data["models"], ["gpt-6.1-sol", "gpt-6-astra"])
# 响应里只有模型 id,绝无密钥字样。
self.assertNotIn("Bearer", json.dumps(data))
self.assertNotIn("XCM_API_KEY", json.dumps(data))
def test_relay_models_requires_auth(self):
status, _ = self._req("GET", "/relay/models")
self.assertEqual(status, 401)
def test_relay_models_failure_is_502(self):
from cloud.relay import RelayError
self.server.relay_fetcher = lambda u, h, t: (_ for _ in ()).throw(
RelayError("relay down"))
_reset_cache_for_tests()
status, data = self._req("GET", "/relay/models", key=self.owner_key)
self.assertEqual(status, 502)
self.assertIn("relay down", data["error"])
def test_approval_mapping(self):
self.assertEqual(_approval_cli_args("ask"),
["--approval-policy", "untrusted"])
self.assertEqual(_approval_cli_args("auto"),
["--approval-policy", "never"])
self.assertEqual(_approval_cli_args("full"),
["--approval-policy", "never",
"--sandbox", "danger-full-access"])
self.assertEqual(_approval_cli_args(None), [])
self.assertEqual(_approval_cli_args("bogus"), [])
def test_local_dispatch_command_carries_options(self):
seen = []
def fake_runner(cmd):
seen.append(cmd)
return SimpleNamespace(returncode=0, stdout="{}")
ex = LocalExecutor("/fake/host.py", config_path="/fake/cfg.json",
runner=fake_runner)
ex.dispatch_task({"id": "t1", "prompt": "hi",
"model": "gpt-6-astra", "effort": "max",
"approval": "full"})
cmd = seen[0]
# 顶层全局参数必须在子命令之前。
self.assertLess(cmd.index("--model"), cmd.index("dispatch"))
self.assertLess(cmd.index("--effort"), cmd.index("dispatch"))
self.assertLess(cmd.index("--approval-policy"),
cmd.index("dispatch"))
self.assertLess(cmd.index("--sandbox"), cmd.index("dispatch"))
self.assertEqual(cmd[cmd.index("--model") + 1], "gpt-6-astra")
self.assertEqual(cmd[cmd.index("--effort") + 1], "max")
self.assertEqual(cmd[cmd.index("--approval-policy") + 1], "never")
self.assertEqual(cmd[cmd.index("--sandbox") + 1],
"danger-full-access")
def test_local_dispatch_without_options_passes_nothing(self):
seen = []
def fake_runner(cmd):
seen.append(cmd)
return SimpleNamespace(returncode=0, stdout="{}")
ex = LocalExecutor("/fake/host.py", runner=fake_runner)
ex.dispatch_task({"id": "t1", "prompt": "hi"})
cmd = seen[0]
self.assertNotIn("--effort", cmd)
self.assertNotIn("--approval-policy", cmd)
self.assertNotIn("--sandbox", cmd)
def test_docker_inner_command_carries_options(self):
ex = DockerExecutor(config_path="/fake/cfg.json")
cmd = ex._inner_host_command(
"dispatch", "--name", "x", "--task", "hi",
model="gpt-6-astra", effort="high", approval="ask")
self.assertLess(cmd.index("--model"), cmd.index("dispatch"))
self.assertLess(cmd.index("--effort"), cmd.index("dispatch"))
self.assertEqual(cmd[cmd.index("--approval-policy") + 1],
"untrusted")
self.assertNotIn("--sandbox", cmd)
if __name__ == "__main__":
unittest.main()
+8 -1
View File
@@ -225,6 +225,10 @@ def _task_from_row(row):
"goal": _col("goal"),
"verify_command": _col("verify_command"),
"goal_max_rounds": _col("goal_max_rounds"),
# 任务级选项:老行缺列时 None = 跟随配置(见 server 侧)。
"model": _col("model"),
"effort": _col("effort"),
"approval": _col("approval"),
}
@@ -539,7 +543,10 @@ class Worker:
"pr_number": "INTEGER", "goal": "TEXT",
"verify_command": "TEXT",
"goal_max_rounds": "INTEGER",
"worktree_host": "TEXT"}.items():
"worktree_host": "TEXT",
"model": "TEXT",
"effort": "TEXT",
"approval": "TEXT"}.items():
if name not in cols:
try:
conn.execute(