diff --git a/cloud/README.md b/cloud/README.md index 80edb28..df27b73 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -1017,6 +1017,50 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳 否则会把 `/admin/sweep`、`/admin/backup` 也暴露到公网）， 否则公网页面调配额接口会落到 28080 而不是控制面。 +## sub2api 网关（每用户独立网关账号） + +- 开关：`CLOUD_GATEWAY`（默认 `xcmapi`；设为 `sub2api` 即切流）。 + 切流前 xcmapi 保持现网；切流本身只是改环境变量 + 重启服务， + 不动任务数据。`SUB2API_BASE_URL`（网关 origin，如 + `http://172.17.0.1:28084`，桥接 IP 供宿主与容器同达）、 + `SUB2API_ADMIN_EMAIL` / `SUB2API_ADMIN_PASSWORD`（系统管理员 + 账号）、`SUB2API_GROUP_NAME`（默认 `codex-system`；或用 + `SUB2API_GROUP_ID` 直接指定组 id）是开通必需的配置。 +- 开通：OAuth 建号时 best-effort 调管理 API 开通（失败记 + `sub2api_provision_failed` 审计，建号本身不失败）；任务开跑 + 前 worker 按 `created_by` 幂等开通（`user_gateway_keys` 表： + user_id → sub2api_user_id、api_key、refresh_token），失败则 + 任务直接 failed，绝不跑错网关。协调员子任务无 created_by 时 + 跟父任务属主走。key 明文存 PG，与 GitHub App 私钥同一信任 + 边界（库只在宿主 loopback 可达）；at-rest 加密是后续项。 + 开通同时授初始余额：`CLOUD_SUB2API_OPENING_BALANCE_USD` + （默认 10 美元，显式配 0 表示不授；非法/负值回落默认）—— + 网关鉴权层在余额 ≤ 0 时直接 403，不授余额的 key 立刻不可 + 用；授余额失败时 key 与账号一并清理再抛错，不留孤儿。 +- 注入：worker 在三条执行路径（Local popen、Local 同步、Docker + 同步）都注入任务级 env：`SUB2API_API_KEY`（属主的 key）、 + `SUB2API_BASE_URL`，并把网关 host 追加进 NO_PROXY（否则 + codex 经 egress 代理访问内网网关会被挡）。host.py 以 `-c` + 注入 provider（`model_provider=sub2api` + provider 表，不改 + 任何租户的 `config.toml`，多网关任务同项目并发互不干扰）， + 未配 `SUB2API_BASE_URL` 时 loud fail。#86 的 shell exclude + 模式已覆盖 `*_KEY`，agent shell 看不到该 key。 +- Docker 密钥通道：`SUB2API_API_KEY` 在 docker 命令里只传名 + （`-e SUB2API_API_KEY`），值由 docker CLI 子进程环境提供 + （`_docker_cli_env`），绝不进 argv（ps/docker inspect 可见 + argv）；runner 不支持 env 时 loud fail。其它 extra_env 仍按 + `-e NAME=VALUE`（非密钥变量）。 +- 模型目录：`/relay/models` 在 sub2api 下返回调用者自己的网关 + 模型（key 绑定的组决定）；无 key 时返回空列表 + + `provisioning: pending`，不在页面加载时悄悄开通。任务未指定 + 模型时：`CLOUD_SUB2API_DEFAULT_MODEL` 优先，否则取该用户 + 网关模型首个；组里无模型时任务 failed（ loud fail，不拿 + xcmapi 的模型名去撞网关）。sub2api 模型目前无价目表，成本 + 走无价 token 上限那道护栏（见上节），报表记 unpriced。 +- 诚实边界：本批只做到“网关可配置 + 任务注入 + 模型目录走 + 网关”；真模型调用实弹需上游账号挂到组里后才做（切流前置 + 条件），fallback 模型名在 sub2api 下须配网关侧的名字。 + ## 运维批（本轮新增：部署、恢复演练、监控、心跳、多用户验证） 细则在 `deploy/README.md`，这里只记代码侧的接口与默认值： diff --git a/cloud/sandbox.py b/cloud/sandbox.py index dd856fd..d690da7 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -130,9 +130,20 @@ def _truncate_stdout(text): return str(text)[:MAX_STDOUT_CHARS] -def _default_runner(cmd): - """subprocess.run 的薄封装：捕获输出、文本模式。""" - return subprocess.run(cmd, capture_output=True, text=True) +def _default_runner(cmd, env=None): + """subprocess.run 的薄封装：捕获输出、文本模式。 + + env 非空时以它为子进程环境（worker 注入任务级密钥时用）； + 不传则继承当前进程环境，与旧行为一致。 + """ + if env is None: + return subprocess.run(cmd, capture_output=True, text=True) + return subprocess.run(cmd, capture_output=True, text=True, env=env) + + +# docker -e 只传名不传值的密钥类变量：值由 CLI 子进程环境提供， +# 绝不进 docker argv（见 build_run_command 与 _docker_cli_env）。 +_DOCKER_SECRET_ENV_NAMES = frozenset({"SUB2API_API_KEY"}) def _approval_cli_args(approval): @@ -190,13 +201,15 @@ class LocalExecutor(Executor): self.model = model def _base_command(self, model=None, effort=None, approval=None, - sandbox=None): + sandbox=None, model_provider=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 model_provider: + cmd += ["--model-provider", model_provider] if effort: cmd += ["--effort", effort] cmd += _approval_cli_args(approval) @@ -207,6 +220,23 @@ class LocalExecutor(Executor): cmd += ["--sandbox", sandbox] return cmd + def _run_local(self, cmd, extra_env): + """本地 runner 调用：extra_env 非空时合并进子进程环境。 + + 假 runner 只接受 cmd 时 TypeError：extra_env 是任务级密钥 + （如 sub2api key），丢了它跑任务等于跑错网关，必须 loud + fail，绝不静默降级成无 key 运行。 + """ + if not extra_env: + return self.runner(cmd) + merged = dict(os.environ) + merged.update(extra_env) + try: + return self.runner(cmd, env=merged) + except TypeError: + raise RuntimeError( + "本地 runner 不支持 env，无法注入任务级环境变量") + def _command(self, task_id): cmd = self._base_command() cmd += ["status", "--task", str(task_id)] @@ -289,7 +319,8 @@ class LocalExecutor(Executor): raw = str(task.get("id")) return re.sub(r"[^A-Za-z0-9_.-]", "-", raw) - def dispatch_task(self, task, seed_dir=None) -> dict: + def dispatch_task(self, task, seed_dir=None, extra_env=None, + model_provider=None) -> dict: """调 host.py dispatch：建 host 任务并跑一轮 turn。 命令：host.py [--config CFG] [--model M] dispatch --name @@ -311,7 +342,8 @@ class LocalExecutor(Executor): effort=task.get("effort"), approval=task.get("approval"), sandbox=("read-only" - if task.get("plan") else None)) + if task.get("plan") else None), + model_provider=model_provider) cmd += ["dispatch", "--name", self._host_name(task), "--task", task["prompt"]] repo = task.get("repo") or self.repo @@ -320,7 +352,7 @@ class LocalExecutor(Executor): if seed_dir: cmd += ["--seed-dir", seed_dir] return self._result_from_runner( - task["id"], self.runner(cmd)) + task["id"], self._run_local(cmd, extra_env)) def fork_task(self, task, source_id, timeout=None, extra_mounts=None, network=None, extra_env=None) -> dict: @@ -343,7 +375,7 @@ class LocalExecutor(Executor): return self._result_from_runner( task["id"], self.runner(cmd)) - def run_task(self, task) -> dict: + def run_task(self, task, extra_env=None, model_provider=None) -> dict: """调 host.py run：在已有 host 任务上再跑一轮 turn。 命令：host.py [--config CFG] [--model M] run --task @@ -359,13 +391,15 @@ class LocalExecutor(Executor): effort=task.get("effort"), approval=task.get("approval"), sandbox=("read-only" - if task.get("plan") else None)) + if task.get("plan") else None), + model_provider=model_provider) cmd += ["run", "--task", self._host_name(task), "--prompt", task["prompt"]] return self._result_from_runner( - task["id"], self.runner(cmd)) + task["id"], self._run_local(cmd, extra_env)) - def goal_task(self, task, seed_dir=None) -> dict: + def goal_task(self, task, seed_dir=None, extra_env=None, + model_provider=None) -> dict: """调 host.py goal：目标循环（turn→验测→喂回重试）。 命令：host.py [--config CFG] [--model M] goal --name @@ -391,7 +425,8 @@ class LocalExecutor(Executor): rounds = 3 cmd = self._base_command(model=task.get("model"), effort=task.get("effort"), - approval=task.get("approval")) + approval=task.get("approval"), + model_provider=model_provider) cmd += ["goal", "--name", self._host_name(task), "--task", goal_text] if verify: @@ -403,9 +438,10 @@ class LocalExecutor(Executor): if seed_dir: cmd += ["--seed-dir", seed_dir] return self._goal_result_from_runner( - task["id"], self.runner(cmd)) + task["id"], self._run_local(cmd, extra_env)) - def goal_resume_task(self, task, goal_text) -> dict: + def goal_resume_task(self, task, goal_text, extra_env=None, + model_provider=None) -> dict: """调 host.py goal-resume --goal-text：在旧线程上开新目标 循环（旧任务接续 goal 模式，云端 /continue 带 goal_mode=1 时走这里）。命令：host.py goal-resume --task @@ -419,12 +455,13 @@ class LocalExecutor(Executor): raise ValueError("goal-resume 缺少目标文本") cmd = self._base_command(model=task.get("model"), effort=task.get("effort"), - approval=task.get("approval")) + approval=task.get("approval"), + model_provider=model_provider) cmd += ["goal-resume", "--task", self._host_name(task), "--goal-text", goal_text, "--max-rounds", "0"] return self._goal_result_from_runner( - task["id"], self.runner(cmd)) + task["id"], self._run_local(cmd, extra_env)) def _goal_result_from_runner(self, task_id, result): """把 host.py goal 的 runner 返回值解析成 goal 结果 dict。 @@ -471,7 +508,8 @@ class LocalExecutor(Executor): "usage": _parse_goal_usage(stdout_full), } - def popen_task(self, task, kind="dispatch", seed_dir=None): + def popen_task(self, task, kind="dispatch", seed_dir=None, + extra_env=None, model_provider=None): """进程组启动 host.py 子进程，返回 Popen（供真取消用）。 start_new_session=True 让子进程自成进程组，worker 取消时 @@ -494,7 +532,8 @@ class LocalExecutor(Executor): effort=task.get("effort"), approval=task.get("approval"), sandbox=("read-only" - if task.get("plan") else None)) + if task.get("plan") else None), + model_provider=model_provider) if kind == "run": cmd += ["run", "--task", self._host_name(task), "--prompt", task["prompt"]] @@ -531,6 +570,12 @@ class LocalExecutor(Executor): cmd += ["--repo", repo] if seed_dir: cmd += ["--seed-dir", seed_dir] + if extra_env: + merged = dict(os.environ) + merged.update(extra_env) + return subprocess.Popen(cmd, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True, + start_new_session=True, env=merged) return subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True) @@ -723,7 +768,9 @@ class DockerExecutor(Executor): -v :/work -v <代码目录>:/app:ro -w /work [-e ] [-e ...] <镜像> <容器内命令...>。 -e 只传名字，不带 =值（env_key）；extra_env 的 -e 带值， - 只用于非密钥变量（如 egress 代理地址）。 + 只用于非密钥变量（如 egress 代理地址）；密钥类变量 + （见 _DOCKER_SECRET_ENV_NAMES）只传名，值由 CLI 子进程 + 环境提供，绝不进 argv。 extra_mounts 给定时替换构造时的静态挂载（多租户隔离 用：worker 按项目算出挂载后每次调用传入）；不给时用 静态挂载。network 给定时容器进该 Docker 网络（多租户 @@ -767,13 +814,35 @@ class DockerExecutor(Executor): if extra_env: merged_env.update(extra_env) for name, value in sorted(merged_env.items()): - cmd += ["-e", "%s=%s" % (name, value)] + # 密钥类变量只传名不传值：docker 从 CLI 子进程环境里 + # 取值（见 _docker_cli_env），值绝不进 docker argv + #（ps/docker inspect 可见 argv，带值等于明文泄露）。 + if name in _DOCKER_SECRET_ENV_NAMES: + cmd += ["-e", name] + else: + cmd += ["-e", "%s=%s" % (name, value)] cmd += [self.image] cmd += list(inner_cmd) return cmd + def _docker_cli_env(self, extra_env): + """docker CLI 子进程环境：extra_env 含密钥名时才需要。 + + 返回合并后的 env（供 -e <名> 按名解析）；不含密钥名时 + 返回 None，保持旧行为（CLI 继承 worker 环境）。 + """ + if not extra_env: + return None + if not any(name in _DOCKER_SECRET_ENV_NAMES + for name in extra_env): + return None + merged = dict(os.environ) + merged.update(extra_env) + return merged + def _inner_host_command(self, *args, model=None, effort=None, - approval=None, sandbox=None): + approval=None, sandbox=None, + model_provider=None): """容器内 host.py 命令：python3 /app/host.py [args...]。 config_path 若已给，容器内用 /app/ 指代它—— @@ -792,6 +861,8 @@ class DockerExecutor(Executor): effective = model or self.model if effective: cmd += ["--model", effective] + if model_provider: + cmd += ["--model-provider", model_provider] if effort: cmd += ["--effort", effort] cmd += _approval_cli_args(approval) @@ -801,19 +872,31 @@ class DockerExecutor(Executor): return cmd # -- runner 调用 / 清理 -------------------------------------------- - def _call_runner(self, cmd, timeout=None): + def _call_runner(self, cmd, timeout=None, env=None): """调 runner：优先传 timeout，假 runner 只接受 cmd 时回退。 timeout 参数只影响本次调用（并发下两个任务各用各的 超时）；不传时用实例的 self.timeout。绝不能在调用方 临时改 self.timeout 再恢复，那在并发下会互相覆盖。 + env 非空时是 docker CLI 子进程的环境（供 -e <名> 按名 + 解析密钥用）：runner 不接受 env 时 loud fail，绝不静默 + 丢掉密钥跑任务（那样容器内 key 缺失，任务必失败还难查）。 """ + if env is None: + try: + return self.runner( + cmd, + timeout=self.timeout if timeout is None else timeout) + except TypeError: + return self.runner(cmd) try: return self.runner( cmd, - timeout=self.timeout if timeout is None else timeout) + timeout=self.timeout if timeout is None else timeout, + env=env) except TypeError: - return self.runner(cmd) + raise RuntimeError( + "docker runner 不支持 env，无法按名注入任务密钥") def list_containers(self, prefix="codex-"): """列出名字以 prefix 开头的容器（默认 codex-），失败返回 []。 @@ -901,7 +984,9 @@ class DockerExecutor(Executor): network=network, extra_env=extra_env) try: - result = self._call_runner(cmd, timeout=timeout) + result = self._call_runner( + cmd, timeout=timeout, + env=self._docker_cli_env(extra_env)) except Exception as exc: if exc.__class__.__name__ == "TimeoutExpired": self.kill_container(task) @@ -926,7 +1011,7 @@ class DockerExecutor(Executor): def dispatch_task(self, task, timeout=None, extra_mounts=None, network=None, extra_env=None, - seed_host_dir=None) -> dict: + seed_host_dir=None, model_provider=None) -> dict: """容器内调 host.py dispatch：建 host 任务并跑一轮 turn。 seed_host_dir 非空时把暂存目录只读挂载进容器 @@ -941,7 +1026,8 @@ class DockerExecutor(Executor): "dispatch", "--name", self._host_name(task), "--task", task["prompt"], model=task.get("model"), effort=task.get("effort"), approval=task.get("approval"), - sandbox=("read-only" if task.get("plan") else None)) + sandbox=("read-only" if task.get("plan") else None), + model_provider=model_provider) repo = task.get("repo") if repo: inner += ["--repo", repo] @@ -972,7 +1058,8 @@ class DockerExecutor(Executor): extra_env=extra_env) def run_task(self, task, timeout=None, extra_mounts=None, - network=None, extra_env=None) -> dict: + network=None, extra_env=None, + model_provider=None) -> dict: """容器内调 host.py run：在已有 host 任务上再跑一轮 turn。""" if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") @@ -982,7 +1069,8 @@ class DockerExecutor(Executor): "run", "--task", self._host_name(task), "--prompt", task["prompt"], model=task.get("model"), effort=task.get("effort"), approval=task.get("approval"), - sandbox=("read-only" if task.get("plan") else None)) + sandbox=("read-only" if task.get("plan") else None), + model_provider=model_provider) return self._run_in_container(task, inner, timeout=timeout, extra_mounts=extra_mounts, network=network, @@ -990,7 +1078,7 @@ class DockerExecutor(Executor): def goal_task(self, task, timeout=None, extra_mounts=None, network=None, extra_env=None, - seed_host_dir=None) -> dict: + seed_host_dir=None, model_provider=None) -> dict: """容器内调 host.py goal：目标循环（turn→验测→喂回重试）。 seed_host_dir 语义同 dispatch_task（只读挂载 + 内层 @@ -1020,7 +1108,8 @@ class DockerExecutor(Executor): inner_args += ["--max-rounds", str(rounds)] inner = self._inner_host_command( *inner_args, model=task.get("model"), - effort=task.get("effort"), approval=task.get("approval")) + effort=task.get("effort"), approval=task.get("approval"), + model_provider=model_provider) repo = task.get("repo") if repo: inner += ["--repo", repo] @@ -1033,7 +1122,9 @@ class DockerExecutor(Executor): network=network, extra_env=extra_env) try: - result = self._call_runner(cmd, timeout=timeout) + result = self._call_runner( + cmd, timeout=timeout, + env=self._docker_cli_env(extra_env)) except Exception as exc: if exc.__class__.__name__ == "TimeoutExpired": self.kill_container(task) @@ -1092,7 +1183,7 @@ class DockerExecutor(Executor): def goal_resume_task(self, task, goal_text, timeout=None, extra_mounts=None, network=None, - extra_env=None) -> dict: + extra_env=None, model_provider=None) -> dict: """容器内调 host.py goal-resume --goal-text：在旧线程上开 新目标循环（旧任务接续 goal 模式）。解析与 goal_task 同款 （完整 stdout 结论、存库截尾），但没有“已存在”分支： @@ -1107,13 +1198,16 @@ class DockerExecutor(Executor): "goal-resume", "--task", self._host_name(task), "--goal-text", goal_text, "--max-rounds", "0", model=task.get("model"), effort=task.get("effort"), - approval=task.get("approval")) + approval=task.get("approval"), + model_provider=model_provider) cmd = self.build_run_command(task, inner, extra_mounts=extra_mounts, network=network, extra_env=extra_env) try: - result = self._call_runner(cmd, timeout=timeout) + result = self._call_runner( + cmd, timeout=timeout, + env=self._docker_cli_env(extra_env)) except Exception as exc: if exc.__class__.__name__ == "TimeoutExpired": self.kill_container(task) diff --git a/cloud/server.py b/cloud/server.py index 10b71c6..e80de04 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -113,6 +113,12 @@ from cloud.oauth import ( ) from cloud.relay import RelayError as _RelayError from cloud.relay import get_relay_models as _get_relay_models +from cloud.sub2api import ensure_gateway_keys_table as _ensure_gateway_keys +from cloud.sub2api import ensure_user_gateway_key as _ensure_sub2api_key +from cloud.sub2api import gateway_name as _gateway_name +from cloud.sub2api import gateway_origin as _gateway_origin +from cloud.sub2api import get_user_gateway_key as _get_sub2api_key +from cloud.sub2api import list_gateway_models as _list_sub2api_models from cloud.repos import validate_repo_url from cloud.push import ( create_subscription as _push_create, @@ -1052,6 +1058,7 @@ def init_db(db_path): _ensure_followups_table(conn) _ensure_environments_table(conn) _ensure_quotas_table(conn) + _ensure_gateway_keys(conn) _ensure_uploads_tables(conn) conn.commit() return @@ -2340,6 +2347,33 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): # 现回，只把模型 id 交给调用者。key 绝不进响应、日志、 # 缓存（见 cloud/relay.py）。单测可经 server.relay_fetcher # 注入假取数函数；中转挂了或未配 key 时 502，不编造列表。 + # 网关为 sub2api 时：模型目录因用户而异（key 绑定的组 + # 决定），取调用者自己的网关 key 现查；无 key 时返回空 + # 列表 + provisioning 标记（控制台据此提示开通中），不 + # 在页面加载时悄悄开通（开通在建号与任务时发生）。 + if _gateway_name() == "sub2api": + conn = self._db() + try: + row = _get_sub2api_key(conn, user_id) + try: + conn.commit() + except Exception: + pass + finally: + conn.close() + if row is None or not row.get("api_key"): + self._send_json(200, {"models": [], "gateway": "sub2api", + "provisioning": "pending"}) + return + try: + models = _list_sub2api_models(_gateway_origin(), + row["api_key"]) + except Exception as exc: + self._send_json(502, {"error": str(exc)}) + return + self._send_json(200, {"models": models, "gateway": "sub2api", + "cached": False, "updated_at": None}) + return fetcher = getattr(self.server, "relay_fetcher", None) try: models, cached, updated_at = _get_relay_models(fetcher=fetcher) @@ -2347,7 +2381,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(502, {"error": str(exc)}) return self._send_json(200, { - "models": models, "cached": cached, "updated_at": updated_at}) + "models": models, "cached": cached, "updated_at": updated_at, + "gateway": "xcmapi"}) def _models_overview(self, user_id): # 优先级列表来自环境变量 MODELS_PRIORITY（逗号分隔），未设时 @@ -4505,6 +4540,14 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): auth.audit(local_id, "", "quota_default", "quota_usd_micros=%d" % _default_quota_usd_micros()) + if created and _gateway_origin() is not None: + # sub2api 开通（best-effort）：建号不因网关开通失败 + # 而失败，任务时 _ensure_gateway 会懒开通兜底。 + try: + _ensure_sub2api_key(auth.conn, local_id) + except Exception as exc: + auth.audit(local_id, "", "sub2api_provision_failed", + str(exc)[:200]) if not created: auth.audit(local_id, "", "api_key_issued", provider) return local_id, created, api_key_plain, rotated, identity, False diff --git a/cloud/sub2api.py b/cloud/sub2api.py new file mode 100644 index 0000000..d6a51b6 --- /dev/null +++ b/cloud/sub2api.py @@ -0,0 +1,357 @@ +"""sub2api 网关接入：系统用管理 API 为每位用户建号发 key。 + +设计（2026-10-27 独立部署 sub2api 后）： +- 系统侧只用管理 API 管用户：管理员登录 -> 建 sub2api 账号 -> + 以该账号登录 -> 发 key -> key 存我方库，任务注入该用户的 key。 +- key 创建是用户侧端点 POST /api/v1/keys（不是 /api/v1/api-keys， + 后者只有管理员改组）；管理员不能代用户建/删 key，所以 refresh_token + 也要存下来：以后轮换/吊销 key 时靠它换用户 access token。 +- key 明文存 PG：与 github_app_config 的 App 私钥同一信任边界 + （库只在宿主 loopback 可达，服务端 env 同机）；at-rest 加密是 + 后续项，不在本批假装已做。 +- 零第三方依赖，只用标准库；HTTP 可注入 request 可调用对象， + 便于单测（见 cloud/tests/test_sub2api.py）。 +""" + +import json +import os +import secrets +import time +import urllib.error +import urllib.request + +GATEWAY_TABLE_SQL = ( + "CREATE TABLE IF NOT EXISTS user_gateway_keys(" + "user_id TEXT PRIMARY KEY, " + "sub2api_user_id BIGINT NOT NULL, " + "api_key TEXT NOT NULL, " + "refresh_token TEXT, " + "created_at DOUBLE PRECISION, " + "updated_at DOUBLE PRECISION)") + + +class Sub2apiError(RuntimeError): + def __init__(self, message, status=None): + super().__init__(message) + self.status = status + + +def _default_request(method, url, body=None, token=None): + """urllib 实现的 request：返回 (status, parsed_json_dict)。""" + headers = {"Content-Type": "application/json"} + if token: + headers["Authorization"] = "Bearer " + token + req = urllib.request.Request( + url, method=method, + data=json.dumps(body).encode() if body is not None else None, + headers=headers) + try: + with urllib.request.urlopen(req, timeout=30) as resp: + raw = resp.read() or b"{}" + return resp.status, json.loads(raw) + except urllib.error.HTTPError as e: + try: + data = json.loads(e.read() or b"{}") + except ValueError: + data = {} + return e.code, data + + +def gateway_origin(): + """sub2api origin（不带 /v1）：未配则返回 None。""" + raw = os.environ.get("SUB2API_BASE_URL", "").strip() + return raw.rstrip("/") or None + + +def gateway_name(): + raw = os.environ.get("CLOUD_GATEWAY", "xcmapi").strip().lower() + return raw if raw in ("xcmapi", "sub2api") else "xcmapi" + + +def gateway_group_name(): + return os.environ.get("SUB2API_GROUP_NAME", "codex-system").strip() \ + or "codex-system" + + +def opening_balance(): + """开通时授的初始余额（美元）：默认 10，显式值优先。 + + 新用户余额为 0 时网关鉴权层直接 403（余额门在任何上游路由 + 之前），所以开通必须同时授余额，否则发出的 key 立刻不可用。 + 显式配 0 表示不授（调用方自理）；非法/负值回落默认 10。 + """ + raw = os.environ.get( + "CLOUD_SUB2API_OPENING_BALANCE_USD", "").strip() + if not raw: + return 10.0 + try: + value = float(raw) + except ValueError: + return 10.0 + return value if value >= 0 else 10.0 + + +def ensure_gateway_keys_table(conn): + conn.execute(GATEWAY_TABLE_SQL) + + +def get_user_gateway_key(conn, user_id): + """只读：返回该用户的网关行 dict 或 None，绝不触发开通。""" + ensure_gateway_keys_table(conn) + cur = conn.execute( + "SELECT user_id, sub2api_user_id, api_key, refresh_token " + "FROM user_gateway_keys WHERE user_id = ?", + (user_id,)) + row = cur.fetchone() + if row is None: + return None + try: + keys = ("user_id", "sub2api_user_id", "api_key", "refresh_token") + return dict(zip(keys, row)) + except (TypeError, ValueError): + return None + + +class Sub2apiAdmin: + """管理员会话：登录、建/删用户、查组。token 自动刷新一次。""" + + def __init__(self, origin, email, password, request=None): + if not origin: + raise Sub2apiError("SUB2API_BASE_URL 未配置") + if not email or not password: + raise Sub2apiError("SUB2API 管理员邮箱/密码未配置") + self.origin = origin + self.email = email + self.password = password + self.request = request or _default_request + self._token = None + self._refresh = None + self._expires_at = 0 + + @classmethod + def from_env(cls, request=None): + return cls(gateway_origin(), + os.environ.get("SUB2API_ADMIN_EMAIL", ""), + os.environ.get("SUB2API_ADMIN_PASSWORD", ""), + request=request) + + def _login(self): + st, data = self.request( + "POST", self.origin + "/api/v1/auth/login", + {"email": self.email, "password": self.password}) + payload = (data or {}).get("data") or {} + if st != 200 or not payload.get("access_token"): + raise Sub2apiError( + "sub2api 管理员登录失败: %s" + % ((data or {}).get("message") or st), status=st) + self._token = payload["access_token"] + self._refresh = payload.get("refresh_token") + try: + ttl = int(payload.get("expires_in") or 3600) + except (TypeError, ValueError): + ttl = 3600 + self._expires_at = time.time() + max(60, ttl - 60) + + def _refresh_token(self): + if not self._refresh: + self._login() + return + st, data = self.request( + "POST", self.origin + "/api/v1/auth/refresh", + {"refresh_token": self._refresh}) + payload = (data or {}).get("data") or {} + if st == 200 and payload.get("access_token"): + self._token = payload["access_token"] + if payload.get("refresh_token"): + self._refresh = payload["refresh_token"] + try: + ttl = int(payload.get("expires_in") or 3600) + except (TypeError, ValueError): + ttl = 3600 + self._expires_at = time.time() + max(60, ttl - 60) + return + self._login() + + def _call(self, method, path, body=None): + if not self._token or time.time() >= self._expires_at: + self._login() + st, data = self.request(method, self.origin + path, body, + token=self._token) + if st == 401: + self._refresh_token() + st, data = self.request(method, self.origin + path, body, + token=self._token) + return st, data + + def create_user(self, email, password, username): + st, data = self._call("POST", "/api/v1/admin/users", { + "email": email, "password": password, + "username": username, "role": "user"}) + payload = (data or {}).get("data") or {} + if st not in (200, 201) or payload.get("id") is None: + raise Sub2apiError( + "sub2api 建用户失败: %s" + % ((data or {}).get("message") or st), status=st) + return payload["id"] + + def delete_user(self, sub_user_id): + st, _data = self._call( + "DELETE", "/api/v1/admin/users/%s" % sub_user_id) + return st in (200, 204) + + def set_balance(self, sub_user_id, amount): + st, data = self._call( + "POST", "/api/v1/admin/users/%s/balance" % sub_user_id, + {"balance": amount, "operation": "set", + "notes": "codex provisioning opening balance"}) + if st != 200: + raise Sub2apiError( + "sub2api 授余额失败: %s" + % ((data or {}).get("message") or st), status=st) + + def list_groups(self): + st, data = self._call("GET", "/api/v1/admin/groups") + payload = (data or {}).get("data") + if st != 200: + raise Sub2apiError("sub2api 查组失败: %s" % st, status=st) + if isinstance(payload, list): + return payload + if isinstance(payload, dict): + for key in ("items", "list", "groups"): + if isinstance(payload.get(key), list): + return payload[key] + return [] + + def resolve_group_id(self): + raw = os.environ.get("SUB2API_GROUP_ID", "").strip() + if raw: + try: + return int(raw) + except ValueError: + raise Sub2apiError("SUB2API_GROUP_ID 非法: %r" % raw) + want = gateway_group_name() + for grp in self.list_groups(): + if isinstance(grp, dict) and grp.get("name") == want: + if grp.get("id") is not None: + return grp["id"] + raise Sub2apiError("sub2api 组不存在: %s" % want) + + +def user_login(origin, email, password, request=None): + """以 sub2api 用户身份登录：返回 (access_token, refresh_token)。""" + req = request or _default_request + st, data = req("POST", origin + "/api/v1/auth/login", + {"email": email, "password": password}) + payload = (data or {}).get("data") or {} + if st != 200 or not payload.get("access_token"): + raise Sub2apiError( + "sub2api 用户登录失败: %s" + % ((data or {}).get("message") or st), status=st) + return payload["access_token"], payload.get("refresh_token") + + +def user_create_key(origin, user_token, name, group_id, request=None): + """以用户身份发 key：返回 (key_id, key_value)。""" + req = request or _default_request + body = {"name": name} + if group_id is not None: + body["group_id"] = group_id + st, data = req("POST", origin + "/api/v1/keys", body, token=user_token) + payload = (data or {}).get("data") or {} + value = payload.get("key") or payload.get("api_key") \ + or payload.get("value") + if st not in (200, 201) or not value: + raise Sub2apiError( + "sub2api 发 key 失败: %s" + % ((data or {}).get("message") or st), status=st) + return payload.get("id"), value + + +def user_delete_key(origin, user_token, key_id, request=None): + req = request or _default_request + st, _data = req("DELETE", origin + "/api/v1/keys/%s" % key_id, + token=user_token) + return st in (200, 204) + + +def user_refresh_access(origin, refresh_token, request=None): + """refresh_token 换 access_token：返回 (新 access, 新 refresh|None)。""" + req = request or _default_request + st, data = req("POST", origin + "/api/v1/auth/refresh", + {"refresh_token": refresh_token}) + payload = (data or {}).get("data") or {} + if st != 200 or not payload.get("access_token"): + raise Sub2apiError("sub2api refresh 失败: %s" % st, status=st) + return payload["access_token"], payload.get("refresh_token") + + +def list_gateway_models(origin, api_key, request=None): + """网关 /v1/models：返回模型 id 列表（key 无模型时为空列表）。""" + req = request or _default_request + st, data = req("GET", origin + "/v1/models", token=api_key) + if st != 200: + raise Sub2apiError("sub2api 查模型失败: %s" % st, status=st) + payload = (data or {}).get("data", data) + if isinstance(payload, dict): + payload = payload.get("data") or [] + out = [] + if isinstance(payload, list): + for item in payload: + if isinstance(item, dict) and item.get("id"): + out.append(item["id"]) + elif isinstance(item, str): + out.append(item) + return out + + +def ensure_user_gateway_key(conn, user_id, admin=None, request=None): + """幂等开通：已有行直接返回 key；否则建号->登录->发 key->授余额->入库。 + + 入库失败时尽力清理刚建的 key 与账号（避免孤儿 key），再抛原错； + 授余额失败同样走这条清理（余额为 0 的 key 在网关侧直接 403， + 留着没有意义）。调用方负责 conn 的提交语义：本函数在入库后 + commit（开通是独立事务，不跟调用方的业务事务绑在一起）。 + """ + ensure_gateway_keys_table(conn) + row = get_user_gateway_key(conn, user_id) + if row is not None: + return row["api_key"] + origin = gateway_origin() + if not origin: + raise Sub2apiError("SUB2API_BASE_URL 未配置，无法开通网关账号") + admin = admin or Sub2apiAdmin.from_env(request=request) + req = request or _default_request + email = "codex-%s@codex.local" % user_id + username = "codex-" + str(user_id)[:12] + password = secrets.token_hex(16) + sub_id = admin.create_user(email, password, username) + key_id = None + try: + user_token, refresh = user_login(origin, email, password, + request=req) + group_id = admin.resolve_group_id() + key_id, key = user_create_key(origin, user_token, + "codex-task-key", group_id, + request=req) + amount = opening_balance() + if amount > 0: + admin.set_balance(sub_id, amount) + now = time.time() + conn.execute( + "INSERT INTO user_gateway_keys(user_id, sub2api_user_id, " + "api_key, refresh_token, created_at, updated_at) " + "VALUES(?, ?, ?, ?, ?, ?)", + (user_id, sub_id, key, refresh, now, now)) + conn.commit() + except Exception: + if key_id is not None: + try: + user_delete_key(origin, user_token, key_id, request=req) + except Exception: + pass + try: + admin.delete_user(sub_id) + except Exception: + pass + raise + return key diff --git a/cloud/tests/test_sub2api.py b/cloud/tests/test_sub2api.py new file mode 100644 index 0000000..16b5341 --- /dev/null +++ b/cloud/tests/test_sub2api.py @@ -0,0 +1,295 @@ +"""sub2api 网关接入单测：管理 API 客户端 + 幂等开通 + provider 覆盖。 + +不跑真实模型、不联网：HTTP 经 request 可调用对象注入假实现； +开通走内存 SQLite；host/sandbox 侧只断言命令与 env 构造。 +""" + +import os +import sqlite3 +import sys +import unittest +from pathlib import Path + +try: + from cloud import sub2api + from cloud.sub2api import Sub2apiAdmin, ensure_user_gateway_key +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud import sub2api + from cloud.sub2api import Sub2apiAdmin, ensure_user_gateway_key + + +def _conn(): + conn = sqlite3.connect(":memory:") + conn.row_factory = sqlite3.Row + return conn + + +class FakeApi: + """按 (method, path) 录调用、按表回包的假 sub2api。""" + + def __init__(self): + self.calls = [] + self.admin_token = "adm-tok" + self.user_token = "usr-tok" + self.groups = [{"id": 2, "name": "codex-system"}] + self.next_user_id = 10 + self.next_key_id = 20 + self.fail_refresh_once = False + self.balance_calls = [] + self.fail_balance = False + + def __call__(self, method, url, body=None, token=None): + path = url.split("28084", 1)[-1] if "28084" in url else url + self.calls.append((method, path, body, token)) + if path == "/api/v1/auth/login" and body.get("email") == "adm@x": + return 200, {"data": {"access_token": self.admin_token, + "refresh_token": "adm-ref", + "expires_in": 3600}} + if path == "/api/v1/auth/login": + return 200, {"data": {"access_token": self.user_token, + "refresh_token": "usr-ref", + "expires_in": 3600}} + if path == "/api/v1/auth/refresh": + return 200, {"data": {"access_token": "new-tok", + "refresh_token": "new-ref", + "expires_in": 3600}} + if path.startswith("/api/v1/admin/users/") \ + and path.endswith("/balance") and method == "POST": + self.balance_calls.append(body) + if self.fail_balance: + return 400, {"message": "bad balance"} + return 200, {"data": {"balance": body.get("balance")}} + if path == "/api/v1/admin/users" and method == "POST": + self.next_user_id += 1 + return 200, {"data": {"id": self.next_user_id}} + if path == "/api/v1/admin/groups" and method == "GET": + return 200, {"data": self.groups} + if path == "/api/v1/keys" and method == "POST": + self.next_key_id += 1 + return 200, {"data": {"id": self.next_key_id, + "key": "sk-probe-%d" % self.next_key_id}} + if path == "/v1/models" and method == "GET": + return 200, {"data": [{"id": "gpt-5"}, {"id": "claude-x"}]} + return 404, {"message": "not found"} + + +class TestSub2apiAdmin(unittest.TestCase): + def setUp(self): + self.api = FakeApi() + self.admin = Sub2apiAdmin("http://x:28084", "adm@x", "pw", + request=self.api) + + def test_create_user_and_group_lookup(self): + uid = self.admin.create_user("u@x", "pw", "u1") + self.assertEqual(uid, 11) + self.assertEqual(self.admin.resolve_group_id(), 2) + + def test_401_refreshes_and_retries(self): + # 第一次管理调用先 401：注入一个先 401 后放行的 request。 + calls = {"n": 0} + real = self.api + + def flaky(method, url, body=None, token=None): + calls["n"] += 1 + if calls["n"] == 2: + return 401, {"message": "expired"} + return real(method, url, body, token) + + admin = Sub2apiAdmin("http://x:28084", "adm@x", "pw", + request=flaky) + uid = admin.create_user("u@x", "pw", "u1") + self.assertEqual(uid, 11) + paths = [c[1] for c in self.api.calls] + self.assertIn("/api/v1/auth/refresh", paths) + + +class TestEnsureGatewayKey(unittest.TestCase): + def setUp(self): + self.api = FakeApi() + self.admin = Sub2apiAdmin("http://x:28084", "adm@x", "pw", + request=self.api) + self.env = {"SUB2API_BASE_URL": "http://x:28084", + "SUB2API_GROUP_NAME": "codex-system"} + self.old = {k: os.environ.get(k) for k in self.env} + os.environ.update(self.env) + + def tearDown(self): + for k, v in self.old.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + + def test_provision_then_idempotent(self): + conn = _conn() + key = ensure_user_gateway_key(conn, "user-abc", admin=self.admin, + request=self.api) + self.assertTrue(key.startswith("sk-probe-")) + calls_before = len(self.api.calls) + key2 = ensure_user_gateway_key(conn, "user-abc", admin=self.admin, + request=self.api) + self.assertEqual(key, key2) + # 第二次是纯读库，不再调管理 API。 + self.assertEqual(len(self.api.calls), calls_before) + row = conn.execute( + "SELECT sub2api_user_id, refresh_token FROM user_gateway_keys" + " WHERE user_id = ?", ("user-abc",)).fetchone() + self.assertEqual(row[0], 11) + self.assertEqual(row[1], "usr-ref") + + def test_list_models(self): + models = sub2api.list_gateway_models("http://x:28084", "sk-1", + request=self.api) + self.assertEqual(models, ["gpt-5", "claude-x"]) + + def test_provision_grants_opening_balance(self): + conn = _conn() + key = ensure_user_gateway_key(conn, "user-bal", admin=self.admin, + request=self.api) + self.assertTrue(key.startswith("sk-probe-")) + self.assertEqual(len(self.api.balance_calls), 1) + self.assertEqual(self.api.balance_calls[0]["balance"], 10.0) + self.assertEqual(self.api.balance_calls[0]["operation"], "set") + + def test_provision_balance_failure_cleans_up(self): + self.api.fail_balance = True + conn = _conn() + with self.assertRaises(sub2api.Sub2apiError): + ensure_user_gateway_key(conn, "user-fail", admin=self.admin, + request=self.api) + paths = [(m, p) for (m, p, _b, _t) in self.api.calls] + self.assertIn(("DELETE", "/api/v1/keys/21"), paths) + self.assertIn(("DELETE", "/api/v1/admin/users/11"), paths) + row = conn.execute( + "SELECT COUNT(*) FROM user_gateway_keys" + " WHERE user_id = ?", ("user-fail",)).fetchone() + self.assertEqual(row[0], 0) + + def test_provision_zero_opening_balance_skips_grant(self): + old = os.environ.get("CLOUD_SUB2API_OPENING_BALANCE_USD") + os.environ["CLOUD_SUB2API_OPENING_BALANCE_USD"] = "0" + try: + conn = _conn() + ensure_user_gateway_key(conn, "user-zero", admin=self.admin, + request=self.api) + finally: + if old is None: + os.environ.pop("CLOUD_SUB2API_OPENING_BALANCE_USD", None) + else: + os.environ["CLOUD_SUB2API_OPENING_BALANCE_USD"] = old + self.assertEqual(self.api.balance_calls, []) + + +class TestProviderCommand(unittest.TestCase): + def test_server_command_sub2api(self): + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + import host + old = os.environ.get("SUB2API_BASE_URL") + os.environ["SUB2API_BASE_URL"] = "http://172.17.0.1:28084" + try: + cmd = host._server_command("/bin/codex", provider="sub2api") + finally: + if old is None: + os.environ.pop("SUB2API_BASE_URL", None) + else: + os.environ["SUB2API_BASE_URL"] = old + self.assertIn("app-server", cmd[-1]) + joined = " ".join(cmd) + self.assertIn('model_provider="sub2api"', joined) + self.assertIn("http://172.17.0.1:28084/v1", joined) + self.assertIn('env_key="SUB2API_API_KEY"', joined) + + def test_server_command_sub2api_needs_base_url(self): + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + import host + old = os.environ.pop("SUB2API_BASE_URL", None) + try: + with self.assertRaises(RuntimeError): + host._server_command("/bin/codex", provider="sub2api") + finally: + if old is not None: + os.environ["SUB2API_BASE_URL"] = old + + def test_server_command_unknown_provider(self): + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + import host + with self.assertRaises(ValueError): + host._server_command("/bin/codex", provider="nope") + + +class TestSandboxProviderFlag(unittest.TestCase): + def test_base_command_flag(self): + from cloud.sandbox import LocalExecutor, DockerExecutor + local = LocalExecutor("/tmp/host.py") + cmd = local._base_command(model_provider="sub2api") + self.assertIn("--model-provider", cmd) + self.assertIn("sub2api", cmd) + docker = DockerExecutor("/tmp/host.py") + inner = docker._inner_host_command("dispatch", + model_provider="sub2api") + self.assertIn("--model-provider", inner) + + def test_run_local_env_merge_and_loud_fail(self): + from cloud.sandbox import LocalExecutor + seen = {} + + def runner(cmd, env=None): + seen["env"] = env + return object() + + local = LocalExecutor("/tmp/host.py", runner=runner) + local._run_local(["x"], {"SUB2API_API_KEY": "sk-1"}) + self.assertEqual(seen["env"]["SUB2API_API_KEY"], "sk-1") + + def cmd_only(cmd): + return object() + + local2 = LocalExecutor("/tmp/host.py", runner=cmd_only) + with self.assertRaises(RuntimeError): + local2._run_local(["x"], {"SUB2API_API_KEY": "sk-1"}) + + +class TestDockerSecretEnv(unittest.TestCase): + def _docker(self, runner): + from cloud.sandbox import DockerExecutor + ex = DockerExecutor("/tmp/host.py", runner=runner, env_key=None) + ex._check_docker = lambda: None + return ex + + def test_secret_name_only_in_argv(self): + from cloud.sandbox import DockerExecutor + seen = {} + + def runner(cmd, timeout=None, env=None): + seen["cmd"] = cmd + seen["env"] = env + return object() + + ex = self._docker(runner) + task = {"id": "abc", "prompt": "hi"} + cmd = ex.build_run_command( + task, ["python3", "/app/host.py", "status"], + extra_env={"SUB2API_API_KEY": "sk-secret", + "HTTP_PROXY": "http://x:8888"}) + self.assertIn("SUB2API_API_KEY", cmd) + self.assertNotIn("SUB2API_API_KEY=sk-secret", cmd) + self.assertIn("HTTP_PROXY=http://x:8888", cmd) + env = ex._docker_cli_env({"SUB2API_API_KEY": "sk-secret"}) + self.assertEqual(env["SUB2API_API_KEY"], "sk-secret") + self.assertIsNone(ex._docker_cli_env({"HTTP_PROXY": "http://x:1"})) + self.assertIsNone(ex._docker_cli_env(None)) + + def test_call_runner_env_loud_fail(self): + from cloud.sandbox import DockerExecutor + + def cmd_only(cmd, timeout=None): + return object() + + ex = self._docker(cmd_only) + with self.assertRaises(RuntimeError): + ex._call_runner(["docker"], env={"SUB2API_API_KEY": "sk-1"}) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index ec9b12f..296dc57 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -147,6 +147,10 @@ from cloud.models import ModelRouter, is_fallbackable, record_fallback from cloud.peer_messages import deliver_peer_message as _deliver_peer_message from cloud.peer_messages import peer_message_prefix as _peer_prefix from cloud.peer_messages import row_get as _row_get +from cloud.sub2api import ensure_user_gateway_key as _ensure_sub2api_key +from cloud.sub2api import gateway_name as _gateway_name +from cloud.sub2api import gateway_origin as _gateway_origin +from cloud.sub2api import list_gateway_models as _list_gateway_models from cloud.pr import PrSkipped, PrStepError from cloud.pr_exec import (claim_pr_job, execute_task_pr, finish_pr_job) from cloud.queue import budget_exceeded, ready_tasks @@ -209,6 +213,18 @@ def resolve_env_key(env_key=None, codex_config_path=None): return DEFAULT_ENV_KEY +def _merge_env(base, override): + """两份 env 合并：override 优先；任一为空时返回另一份原样 + （保持旧调用在双 None 时传 None，不改执行器行为）。""" + if not base: + return override + if not override: + return base + merged = dict(base) + merged.update(override) + return merged + + def _resolve_model(model=None, config_path=None): """模型名：显式参数 > host 配置文件里的 model > host 默认值。""" if model: @@ -1119,6 +1135,93 @@ class Worker: is _sandbox_default_runner and hasattr(self.executor, "popen_task")) + def _gateway_context(self, task): + """网关注入：xcmapi -> (None, None)；sub2api -> (provider, env)。 + + sub2api 时按任务属主（tasks.created_by）幂等开通并取 key； + 开通失败直接抛（任务失败，不跑错网关）。env 含 + SUB2API_API_KEY、SUB2API_BASE_URL，并把网关 host 追加进 + NO_PROXY/no_proxy（否则 codex 经 egress 代理访问内网网关 + 会被挡；proxy_env 的 NO_PROXY 会被这份 env 覆盖，所以这里 + 直接给超集 localhost,127.0.0.1,<网关 host>）。 + """ + if _gateway_name() != "sub2api": + return None, None + origin = _gateway_origin() + if not origin: + raise RuntimeError("网关为 sub2api 但 SUB2API_BASE_URL 未配置") + owner = task.get("created_by") + if not owner and task.get("parent_task_id"): + # 协调员子任务：属主跟父任务走（子任务行可能无 created_by）。 + try: + conn = _db_connect(self.db_path) + try: + row = conn.execute( + "SELECT created_by FROM tasks WHERE id = ?", + (task["parent_task_id"],)).fetchone() + finally: + conn.close() + if row is not None: + try: + owner = row[0] + except (TypeError, IndexError): + owner = None + except Exception: + owner = None + if not owner: + raise RuntimeError("任务缺少 created_by，无法解析网关属主") + conn = _db_connect(self.db_path) + try: + key = _ensure_sub2api_key(conn, owner) + finally: + conn.close() + host = origin.split("://", 1)[-1].split("/", 1)[0] + host = host.split(":", 1)[0] + no_proxy = "localhost,127.0.0.1" + if host and host not in ("localhost", "127.0.0.1"): + no_proxy += "," + host + env = { + "SUB2API_API_KEY": key, + "SUB2API_BASE_URL": origin, + "NO_PROXY": no_proxy, + "no_proxy": no_proxy, + } + return "sub2api", env + + def _ensure_gateway(self, task): + """任务 dict 上的网关缓存：已算过直接取，否则算一次并存回。 + + 键是 _gateway_provider/_gateway_env（与 _resume 同款下划线 + 私有键约定，不进库）。run_one 算过后 _run_attempt 不再 + 重算；_run_fork 直达 _run_attempt 时在这里补算。 + """ + if "_gateway_provider" in task: + return task["_gateway_provider"], task["_gateway_env"] + provider, env = self._gateway_context(task) + task["_gateway_provider"] = provider + task["_gateway_env"] = env + return provider, env + + def _sub2api_default_model(self, gateway_env): + """sub2api 默认模型：环境变量优先，否则取该用户网关模型首个。 + + 网关模型目录因用户而异（key 绑定的组决定），worker 构造时 + 没有属主，只能到任务时现取；取不到（组里无模型）就 loud + fail，不让任务拿着 xcmapi 的模型名去撞 sub2api。 + """ + raw = os.environ.get("CLOUD_SUB2API_DEFAULT_MODEL", "").strip() + if raw: + return raw + origin = _gateway_origin() + key = (gateway_env or {}).get("SUB2API_API_KEY") + if not origin or not key: + raise RuntimeError("sub2api 默认模型无法解析：缺少网关 key") + models = _list_gateway_models(origin, key) + if not models: + raise RuntimeError( + "sub2api 网关暂无可用模型（组里未挂上游账号？）") + return models[0] + def _task_seed_dir(self, task): """任务附件暂存目录（派发时 seed 用）：无附件返回 None； 有附件行但目录已不在（如终局后已清理）也返回 None—— @@ -1175,6 +1278,7 @@ class Worker: 里）。Docker/同步路径没有这个轮询钩子，只走边界检查。 """ task_id = task["id"] + gateway_provider, gateway_env = self._ensure_gateway(task) if self.process_factory is not None: proc = self.process_factory(task) else: @@ -1182,7 +1286,8 @@ class Worker: # （文件已在 worktree 里）；popen_task 内部同样只在 # dispatch/goal 命令上追加 --seed-dir。 proc = self.executor.popen_task( - task, kind=kind, seed_dir=seed_host_dir) + task, kind=kind, seed_dir=seed_host_dir, + extra_env=gateway_env, model_provider=gateway_provider) self._running_set(task_id, proc=proc) try: self._set_child_pid(task_id, getattr(proc, "pid", None)) @@ -1899,6 +2004,7 @@ class Worker: except (TypeError, ValueError): effective_docker_timeout = self.task_timeout try: + gateway_provider, gateway_env = self._ensure_gateway(task) if self._use_popen(): if goal_mode and task.get("_resume"): popen_kind = "goal-resume" @@ -1944,42 +2050,60 @@ class Worker: timeout=effective_docker_timeout, extra_mounts=tenant_mounts, network=tenant_network, - extra_env=proxy_env) + extra_env=_merge_env(proxy_env, + gateway_env), + model_provider=gateway_provider) else: result = self.executor.goal_task( task, timeout=effective_docker_timeout, extra_mounts=tenant_mounts, network=tenant_network, - extra_env=proxy_env, - seed_host_dir=seed_host_dir) + extra_env=_merge_env(proxy_env, + gateway_env), + seed_host_dir=seed_host_dir, + model_provider=gateway_provider) else: result = self.executor.dispatch_task( task, timeout=effective_docker_timeout, extra_mounts=tenant_mounts, network=tenant_network, - extra_env=proxy_env, - seed_host_dir=seed_host_dir) + extra_env=_merge_env(proxy_env, gateway_env), + seed_host_dir=seed_host_dir, + model_provider=gateway_provider) else: # seed 参数只传给 LocalExecutor：CloudExecutor # 基类与测试假执行器的 goal_task/dispatch_task # 只有 (task) 一个参数，多传 seed_dir 会直接 # TypeError（2026-10-27 实测炸掉 3 个旧测试）； # Docker 路径上面已有 is_docker 守卫，不走这里。 + # 网关任务在非 Local 执行器上无法注入 key 与 + # provider，必须 loud fail，不静默跑错网关。 + if gateway_provider and not isinstance( + self.executor, LocalExecutor): + raise RuntimeError( + "sub2api 网关不支持该执行器：%s" + % type(self.executor).__name__) if goal_mode: if task.get("_resume") and isinstance( self.executor, LocalExecutor): result = self.executor.goal_resume_task( - task, task["prompt"]) + task, task["prompt"], + extra_env=gateway_env, + model_provider=gateway_provider) elif isinstance(self.executor, LocalExecutor): result = self.executor.goal_task( - task, seed_dir=seed_host_dir) + task, seed_dir=seed_host_dir, + extra_env=gateway_env, + model_provider=gateway_provider) else: result = self.executor.goal_task(task) else: if isinstance(self.executor, LocalExecutor): result = self.executor.dispatch_task( - task, seed_dir=seed_host_dir) + task, seed_dir=seed_host_dir, + extra_env=gateway_env, + model_provider=gateway_provider) else: result = self.executor.dispatch_task(task) combined = (result.get("stdout", "") or "") + \ @@ -1993,10 +2117,21 @@ class Worker: if self.tenant_isolation else None, network=tenant_network if self.tenant_isolation else None, - extra_env=proxy_env - if self.egress_proxy_port else None) + extra_env=_merge_env( + proxy_env if self.egress_proxy_port + else None, gateway_env), + model_provider=gateway_provider) else: - result = self.executor.run_task(task) + if gateway_provider and not isinstance( + self.executor, LocalExecutor): + raise RuntimeError( + "sub2api 网关不支持该执行器：%s" + % type(self.executor).__name__) + result = self.executor.run_task( + task, extra_env=gateway_env, + model_provider=gateway_provider) \ + if isinstance(self.executor, LocalExecutor) \ + else self.executor.run_task(task) except Exception as exc: # runner 异常也如实写回 failed result = {"executor": "local", "task_id": task_id, "status": "failed", "stdout": "", "returncode": None, @@ -2445,6 +2580,35 @@ class Worker: # 再兜底）。此前这里无条件写成 worker 主模型，导致控制 # 台选的模型被悄悄覆盖（任务级模型实际不生效）；修掉。 # fallback 时下面再把 task["model"] 换成 fallback 模型。 + # 网关门：sub2api 时先给属主开通网关账号（幂等）并解析默认 + # 模型；失败直接 failed，不进执行器（避免跑错网关或拿着 + # xcmapi 模型名去撞 sub2api）。xcmapi 下 _ensure_gateway + # 直接返回 (None, None)，不做任何事。注意：这里改的是 + # 原 task dict，随后 task = dict(task) 会原样拷走。 + try: + gateway_provider, gateway_env = self._ensure_gateway(task) + if not task.get("model") and gateway_provider == "sub2api": + task["model"] = self._sub2api_default_model(gateway_env) + except Exception as exc: + error = "网关准备失败: %s" % exc + self._set_status(task_id, "failed") + self._write_execution_result( + task_id, error_text=error, actual_model=self.model, + started_at=started_at, finished_at=time.time()) + self._audit(task["project_id"], "worker_gateway_failed", + "%s: %s" % (task_id, error)) + log_event(self.logger, "worker_gateway_failed", + task_id=task_id, project_id=task["project_id"], + status="failed") + self._notify_task_finished( + task_id, task["project_id"], "failed", error=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": self._executor_name(), + "stdout": "", "error": error, "events": [], + "cost_record_id": None} + task = dict(task) if not task.get("model"): task["model"] = self.model diff --git a/host.py b/host.py index 2b006dc..b30abd8 100644 --- a/host.py +++ b/host.py @@ -319,28 +319,49 @@ def _append_trace_marker(state_path, task_name, entry): pass -def _server_command(codex_bin): +def _server_command(codex_bin, provider=None): """app-server 启动命令：-c 必须放在子命令之前。 shell_environment_policy.exclude 只过滤 agent shell 看到的 变量，不影响 server 进程自己的中转调用（env_key 直读进程 env）。 + provider="sub2api" 时再叠加 -c 覆盖：model_provider 与 + provider 表都按任务进程生效，不改任何租户的 config.toml + （多网关任务同项目并发时互不干扰）。网关地址取环境变量 + SUB2API_BASE_URL（origin，不带 /v1）；未配则 loud fail， + 绝不静默跑回默认网关。 """ - return [ + cmd = [ codex_bin, "-c", "shell_environment_policy.exclude=" + json.dumps(SHELL_ENV_EXCLUDE), - "app-server", ] + if provider is not None: + if provider != "sub2api": + raise ValueError("未知模型 provider: %r" % provider) + origin = os.environ.get("SUB2API_BASE_URL", "").strip().rstrip("/") + if not origin: + raise RuntimeError( + "provider=sub2api 但 SUB2API_BASE_URL 未配置") + cmd += [ + "-c", 'model_provider="sub2api"', + "-c", 'model_providers.sub2api.name="Sub2API"', + "-c", 'model_providers.sub2api.base_url="%s/v1"' % origin, + "-c", 'model_providers.sub2api.env_key="SUB2API_API_KEY"', + "-c", 'model_providers.sub2api.wire_api="responses"', + ] + cmd.append("app-server") + return cmd class Server: """单个 app-server 进程，多 thread 并发。""" def __init__(self, codex_bin=None, codex_home=None, approval="deny", - progress_interval=30): + progress_interval=30, model_provider=None): self.codex_bin = codex_bin or default_codex_bin() self.codex_home = codex_home self.approval = approval + self.model_provider = model_provider # 流式进度间隔（秒）：run_turn 等待时每满一个间隔打一次 # 进度；0 表示关闭进度输出，但中断轮询仍每秒进行。 self.progress_interval = progress_interval @@ -396,7 +417,8 @@ class Server: os.makedirs(self.codex_home, exist_ok=True) env["CODEX_HOME"] = self.codex_home self.proc = subprocess.Popen( - _server_command(self.codex_bin), + _server_command(self.codex_bin, + provider=self.model_provider), stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, env=env, ) @@ -1995,6 +2017,9 @@ def main(): help="项目配置文件，默认 host.py 同目录的 host-config.json（若存在）") ap.add_argument("--state", default=None) ap.add_argument("--model", default=None) + ap.add_argument("--model-provider", default=None, + help="模型 provider 覆盖：sub2api 时以 -c 注入网关" + " provider（不改 config.toml）；不传则走配置") ap.add_argument("--effort", default=None) ap.add_argument("--sandbox", default=None) ap.add_argument("--approval-policy", default=None) @@ -2381,7 +2406,8 @@ def main(): return srv = Server(codex_bin=args.codex_bin, codex_home=args.codex_home, approval=args.approval, - progress_interval=args.progress_interval) + progress_interval=args.progress_interval, + model_provider=args.model_provider) # 实时链路：只给跑 turn 的命令设链路文件（dispatch/run/goal/ # delegate）；只读命令不设。名字映射与 worker 派发一致： # dispatch/goal/delegate 用 --name，run 用 --task。