sub2api 网关接入：每用户独立账号 + 任务注入 + 网关可配置

- cloud/sub2api.py：管理 API 客户端（管理员登录/token 刷新/
  建用户/以用户发 key/删用户/查组/网关模型目录）+ 幂等开通
  ensure_user_gateway_key（入库失败尽力清理孤儿 key 与账号）
- worker：网关门（sub2api 时按 created_by 幂等开通，失败任务
  直接 failed；协调员子任务跟父任务属主）+ 默认模型走网关
  目录首个；三条执行路径（Local popen/同步、Docker 同步）注
  入任务级 env 与 --model-provider；NO_PROXY 追加网关 host
- host.py：--model-provider，sub2api 时以 -c 注入 provider 表
  （不改租户 config.toml）；未配 SUB2API_BASE_URL 时 loud fail
- sandbox：Local/Docker 执行器传 --model-provider 与任务 env；
  docker 密钥只传名（-e SUB2API_API_KEY），值由 CLI 子进程
  环境提供，绝不进 argv；runner 不支持 env 时 loud fail
- server：OAuth 建号 best-effort 开通；/relay/models 走调用者
  自己的网关模型目录；init_db 补 user_gateway_keys 表
- 测试：新单测 11 个；全量 cloud 679（4 跳过）、host 65、
  deploy 19 全过
- 诚实边界：真模型调用实弹需上游账号挂组后才做；切流前
  xcmapi 保持现网

cloud/README.md

@@ -1017,6 +1017,46 @@ 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 加密是后续项。
+- 注入：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`，这里只记代码侧的接口与默认值：

cloud/sandbox.py

@@ -130,9 +130,20 @@ def _truncate_stdout(text):
     return str(text)[:MAX_STDOUT_CHARS]
 
 
-def _default_runner(cmd):
+def _default_runner(cmd, env=None):
-    """subprocess.run 的薄封装：捕获输出、文本模式。"""
+    """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 <host 名>
@@ -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 <host 名>
@@ -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 <host 名>
@@ -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 <host 名>
@@ -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 <worktree>:/work -v <代码目录>:/app:ro -w /work
         [-e <env_key名>] [-e <NAME=VALUE>...] <镜像> <容器内命令...>。
         -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()):
+            # 密钥类变量只传名不传值：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/<basename> 指代它——
@@ -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,
+                env=env)
+        except TypeError:
+            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)

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

cloud/sub2api.py

@@ -0,0 +1,325 @@
+"""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 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 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），再抛原错。
+    调用方负责 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)
+        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

cloud/tests/test_sub2api.py

@@ -0,0 +1,250 @@
+"""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
+
+    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 == "/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"])
+
+
+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()

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,
+                                extra_env=_merge_env(proxy_env,
-                                seed_host_dir=seed_host_dir)
+                                                     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,
+                            extra_env=_merge_env(proxy_env, gateway_env),
-                            seed_host_dir=seed_host_dir)
+                            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
+                            extra_env=_merge_env(
-                            if self.egress_proxy_port else None)
+                                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

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。