Merge pull request #68 from openqj/env-console

feat: 命令台云端环境——选、建、管 + 任务绑定环境

cloud/README.md

@@ -782,6 +782,54 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳
   任务仍需 worker 认领执行；`CloudExecutor` 仍是占位符；公网 https
   入口仍需部署侧反向代理 + TLS。
 
+## 云端环境（选、建、管）
+
+环境是项目仓库的一个命名基准检出，不是官方 Codex 的云端环境
+（那是官方产品自己的后端，本控制面没有它的公开调用口，不宣称
+能建它）。本控制面的环境管的是主机/Docker 这边的基准：项目 +
+仓库快照 + 分支 + 执行器，任务选了环境后仍是一任务一 worktree
+（host.py 从环境检出 HEAD 为任务建分支），任务隔离、合并、提 PR
+全都不变。
+
+- API：`POST /environments`（project_id、name 必填，branch 可选
+  空=默认分支，executor 可选 local/docker；建完 status=creating）、
+  `GET /environments?project_id=`（默认排除已归档，
+  `include_archived=1` 才含归档）、`GET /environments/{id}`、
+  `POST /environments/{id}/archive`。写操作（建/归档）要求项目
+  create_task 权限，读要求 read 权限。
+- 创建规则：项目没有 repo_url 时 400（没有仓库就没有环境）；
+  同项目未归档环境名唯一（409，归档过的名字可重用，部分唯一
+  索引保证）；分支名只允许字母数字与 `_ . - /`，不以 `-` 开头、
+  不含 `..`、不以 `.lock` 结尾；executor 只接受 local/docker。
+- 准备：worker 在 `run_all` 每轮开头准备所有 creating 的环境：
+  目录不存在时 `git clone`，存在时校验 git 仓库且 origin 与环境
+  repo_url 一致再 `fetch`，分支非空时 `checkout` 该分支（不存在
+  则失败，绝不自动建分支），成功变 ready、失败变 failed 并写
+  error_text。准备本身幂等，没有 preparing 中间态——worker 中途
+  死了、环境仍是 creating 时下一轮会重新准备；准备期间用户归档
+  了环境时 CAS 写回改不到行，状态保持 archived。
+- 任务绑定：`POST /tasks` 接受可选 `environment_id`，必须是本项目
+  已 ready 的环境（creating/failed/archived 一律 400 并写明状态）；
+  环境 executor 非空且任务 executor 已指定时必须一致，任务
+  executor 未指定时继承环境 executor（认领过滤靠它）；cloud
+  执行器没有环境检出路径，一律拒绝。worker 派发时环境任务不走
+  项目仓库路径，直接把 `task["repo"]` 指到环境检出（租户 Docker
+  为容器内路径 `/var/lib/codex-app-server/envs/<id>`，租户挂载里
+  整目录挂载该项目 envs 目录；非租户为 `repos_dir/envs/<id>`）。
+- 管理：归档幂等（已归档再归档返回 200）；有 queued/running
+  任务正在用环境时归档 409；归档只改状态，不删检出目录（已完成
+  任务的 worktree 独立，检出留着可查，磁盘清理另行手动做）。
+  重试（retry）只改任务状态，environment_id 原样保留；会话分支
+  （fork）不继承 environment_id——分支是 thread/worktree 分叉，
+  不是在环境上重开任务。
+- 命令台：项目卡下方新增“环境”卡（列表/新建/归档，creating→
+  ready 的过程靠 10 秒自动刷新可见，失败原因直接显示在列表里）；
+  新建任务栏新增环境下拉（未 ready 的置灰、不可选），任务列表里
+  带环境任务会标 `[环境 <短 id>]`。公网 nginx 若要让页面调通环境
+  接口，须把 `/environments` 同批代理到双机上游（与 `/console`、
+  `/tasks`、`/projects`、`/workers`、`/health` 同款，否则页面能打开
+  但环境区永远加载失败）。
+
 ## 运维批（本轮新增：部署、恢复演练、监控、心跳、多用户验证）
 
 细则在 `deploy/README.md`，这里只记代码侧的接口与默认值：

cloud/console.html

@@ -118,9 +118,30 @@
 </div>
 </div>
 
+<section class="card">
+<h2>环境 <span class="muted">（选、建、管：当前项目的基准检出）</span></h2>
+<div class="bar">
+  <button id="refreshEnvironments">刷新</button>
+  <span id="envCount" class="muted"></span>
+</div>
+<div id="envList" class="muted">先选项目</div>
+<div class="bar">
+  <input id="newEnvName" type="text" placeholder="新环境名">
+  <input id="newEnvBranch" type="text" placeholder="分支（可选，空=默认分支）">
+  <select id="newEnvExecutor">
+    <option value="">执行器：不限</option>
+    <option value="local">local</option>
+    <option value="docker">docker</option>
+  </select>
+  <button id="createEnvironment">新建环境</button>
+</div>
+<span id="envMsg" class="muted"></span>
+</section>
+
 <section class="card">
 <h2>新建任务</h2>
 <div class="bar">
+  <select id="envSelect"><option value="">环境：不指定（走项目仓库）</option></select>
   <select id="executor">
     <option value="">executor：任意 worker</option>
     <option value="local">local</option>
@@ -300,6 +321,51 @@ async function loadWorkers() {
   } catch (e) { $("#workers").textContent = "加载失败：" + e.message; }
 }
 
+async function loadEnvironments() {
+  var pid = $("#projectSelect").value;
+  var sel = $("#envSelect");
+  var cur = sel.value;
+  if (!pid) {
+    $("#envList").textContent = "先选项目";
+    $("#envCount").textContent = "";
+    sel.innerHTML = "<option value=''>环境：不指定（走项目仓库）</option>";
+    return;
+  }
+  try {
+    var d = await api("/environments?project_id=" + encodeURIComponent(pid), "GET");
+    var envs = d.environments || [];
+    $("#envCount").textContent = envs.length + " 个环境";
+    if (!envs.length) { $("#envList").textContent = "该项目暂无环境（先建一个，或不指定环境直接建任务）"; }
+    else {
+      $("#envList").innerHTML = envs.map(function (e) {
+        var btn = e.status === "archived" ? "" : " <button data-env-archive='" + esc(e.id) + "'>归档</button>";
+        var err = e.status === "failed" && e.error_text ? " / 失败原因：" + esc(e.error_text) : "";
+        return esc(e.name) + "（" + esc(shortId(e.id)) + "） / 分支 " + esc(e.branch || "默认分支") +
+          " / 执行器 " + esc(e.executor || "不限") + " / 状态 " + esc(e.status) + err + btn + "<br>";
+      }).join("");
+      Array.prototype.forEach.call(document.querySelectorAll("[data-env-archive]"), function (b) {
+        b.onclick = async function () {
+          if (!confirm("确定归档该环境吗？归档后新任务不能再选它，进行中的任务不受影响（有进行中任务在用它时归档会被拒绝）。")) return;
+          try { await api("/environments/" + encodeURIComponent(b.getAttribute("data-env-archive")) + "/archive", "POST", {}); loadEnvironments(); }
+          catch (e) { $("#envMsg").innerHTML = "<span class='err'>归档失败：" + esc(e.message) + "</span>"; }
+        };
+      });
+    }
+    // 任务创建的环境下拉：只列未归档环境，未 ready 的置灰（creating
+    // 时可看到它在准备，ready 后才能选）；刷新时尽量保留已选值。
+    sel.innerHTML = "<option value=''>环境：不指定（走项目仓库）</option>";
+    envs.forEach(function (e) {
+      var o = document.createElement("option");
+      o.value = e.id; o.textContent = e.name + "（" + e.status + "）";
+      if (e.status !== "ready") o.disabled = true;
+      sel.appendChild(o);
+    });
+    if (cur && Array.prototype.some.call(sel.options, function (o) { return o.value === cur && !o.disabled; })) sel.value = cur;
+  } catch (e) {
+    $("#envList").innerHTML = "<span class='err'>加载失败：" + esc(e.message) + "</span>";
+  }
+}
+
 async function loadTasks() {
   var pid = $("#projectSelect").value;
   if (!pid) { $("#taskRows").innerHTML = "<tr><td colspan='7' class='muted'>先选项目</td></tr>"; return; }
@@ -312,7 +378,7 @@ async function loadTasks() {
       var pr = t.pr_url ? "<a href='" + esc(t.pr_url) + "' target='_blank' rel='noopener'>#" + esc(t.pr_number) + "</a>" : "";
       return "<tr data-id='" + esc(t.id) + "' class='" + (currentTask && currentTask.id === t.id ? "selected" : "") + "'>" +
         "<td>" + esc(shortId(t.id)) + "</td>" +
-        "<td>" + (t.plan ? "[计划] " : "") + (t.forked_from ? "[分支] " : "") + ((t.goal_mode || t.verify_command) ? "[目标] " : "") + esc(String(t.prompt || "").slice(0, 60)) + "</td>" +
+        "<td>" + (t.plan ? "[计划] " : "") + (t.forked_from ? "[分支] " : "") + ((t.goal_mode || t.verify_command) ? "[目标] " : "") + (t.environment_id ? "[环境 " + esc(shortId(t.environment_id)) + "] " : "") + esc(String(t.prompt || "").slice(0, 60)) + "</td>" +
         "<td class='st-" + esc(t.status) + "'>" + esc(t.status) + "</td>" +
         "<td>" + esc(t.executor || "") + "</td>" +
         "<td>" + esc(t.worktree_host || "") + "</td>" +
@@ -507,7 +573,7 @@ $("#saveKey").onclick = function () {
   loadHealth();
   loadProjects();
   loadWorkers();
-  if ($("#projectSelect").value) loadTasks();
+  if ($("#projectSelect").value) { loadTasks(); loadEnvironments(); }
   if (getKey()) loadRelayModels();
 };
 $("#clearKey").onclick = function () { localStorage.removeItem(KEY_STORE); setKeyState(); };
@@ -515,7 +581,8 @@ $("#refreshHealth").onclick = loadHealth;
 $("#refreshProjects").onclick = loadProjects;
 $("#refreshWorkers").onclick = loadWorkers;
 $("#refreshTasks").onclick = loadTasks;
-$("#projectSelect").onchange = loadTasks;
+$("#refreshEnvironments").onclick = loadEnvironments;
+$("#projectSelect").onchange = function () { loadTasks(); loadEnvironments(); };
 $("#goalMode").onchange = function () { $("#goalHint").style.display = this.checked ? "block" : "none"; };
 $("#createProject").onclick = async function () {
   var name = $("#newProjectName").value.trim();
@@ -530,12 +597,31 @@ $("#createProject").onclick = async function () {
     loadProjects();
   } catch (e) { $("#projectMsg").innerHTML = "<span class='err'>新建失败：" + esc(e.message) + "</span>"; }
 };
+$("#createEnvironment").onclick = async function () {
+  var pid = $("#projectSelect").value;
+  var name = $("#newEnvName").value.trim();
+  if (!pid) { $("#envMsg").textContent = "先选项目"; return; }
+  if (!name) { $("#envMsg").textContent = "环境名必填"; return; }
+  var body = { project_id: pid, name: name };
+  var br = $("#newEnvBranch").value.trim();
+  if (br) body.branch = br;
+  var ex = $("#newEnvExecutor").value;
+  if (ex) body.executor = ex;
+  try {
+    await api("/environments", "POST", body);
+    $("#envMsg").innerHTML = "<span class='ok'>环境已建，worker 准备中（creating → ready，失败会在列表里写原因）</span>";
+    $("#newEnvName").value = ""; $("#newEnvBranch").value = "";
+    loadEnvironments();
+  } catch (e) { $("#envMsg").innerHTML = "<span class='err'>新建失败：" + esc(e.message) + "</span>"; }
+};
 $("#createTask").onclick = async function () {
   var pid = $("#projectSelect").value;
   var prompt = $("#prompt").value.trim();
   if (!pid) { $("#taskMsg").textContent = "先选项目"; return; }
   if (!prompt) { $("#taskMsg").textContent = "prompt 必填"; return; }
   var body = { project_id: pid, prompt: prompt };
+  var envId = $("#envSelect").value;
+  if (envId) body.environment_id = envId;
   var ex = $("#executor").value;
   if (ex) body.executor = ex;
   var model = $("#modelCustom").value.trim() || $("#modelSelect").value;
@@ -602,7 +688,7 @@ loadWorkers();
 // 否则下拉永远只有默认项（此前只在点“保存”时调一次，
 // 2026-10-27 实测：存过 key 后重开页面模型选不了）。
 if (getKey()) loadRelayModels();
-setInterval(function () { if ($("#projectSelect").value) loadTasks(); loadHealth(); }, 10000);
+setInterval(function () { if ($("#projectSelect").value) { loadTasks(); loadEnvironments(); } loadHealth(); }, 10000);
 </script>
 </body>
 </html>

cloud/db.py

@@ -837,6 +837,7 @@ POSTGRES_SCHEMA = (
     "approval TEXT, "
     "plan INTEGER, "
     "forked_from TEXT, "
+    "environment_id TEXT, "
     "UNIQUE(project_id, idempotency_key))",
     "CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status)",
     "CREATE TABLE IF NOT EXISTS task_followups("
@@ -851,6 +852,23 @@ POSTGRES_SCHEMA = (
     "updated_at DOUBLE PRECISION)",
     "CREATE INDEX IF NOT EXISTS idx_followups_task"
     " ON task_followups(task_id, status)",
+    "CREATE TABLE IF NOT EXISTS environments("
+    "id TEXT PRIMARY KEY, "
+    "project_id TEXT, "
+    "name TEXT, "
+    "branch TEXT, "
+    "executor TEXT, "
+    "status TEXT, "
+    "error_text TEXT, "
+    "repo_url TEXT, "
+    "created_by TEXT, "
+    "created_at DOUBLE PRECISION, "
+    "updated_at DOUBLE PRECISION)",
+    "CREATE INDEX IF NOT EXISTS idx_env_project"
+    " ON environments(project_id, status)",
+    "CREATE UNIQUE INDEX IF NOT EXISTS idx_env_name"
+    " ON environments(project_id, name)"
+    " WHERE status != 'archived'",
     "CREATE TABLE IF NOT EXISTS summaries("
     "id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, "
     "project_id TEXT, "
@@ -931,7 +949,8 @@ def init_postgres_schema(conn):
     for _name, _type in (("executor", "TEXT"), ("pr_url", "TEXT"),
                          ("pr_number", "INTEGER"),
                          ("worktree_host", "TEXT"),
-                         ("goal_mode", "INTEGER")):
+                         ("goal_mode", "INTEGER"),
+                         ("environment_id", "TEXT")):
         if _name not in tcols:
             conn.execute(
                 "ALTER TABLE tasks ADD COLUMN %s %s" % (_name, _type))

cloud/server.py

@@ -452,6 +452,10 @@ def migrate_tasks_table(conn):
         # read-only 沙箱；forked_from 记源任务 id（非分支为 NULL）。
         "plan": "INTEGER",
         "forked_from": "TEXT",
+        # 环境：任务绑定的环境 id（NULL = 不指定环境，走原有
+        # 项目仓库路径）；环境本身是项目仓库的一个命名基准检出
+        # （分支 + 执行器），任务 worktree 从它建，不改任务隔离。
+        "environment_id": "TEXT",
     }
     for name, coltype in new_cols.items():
         if name not in cols:
@@ -510,6 +514,85 @@ def _followup_to_dict(row):
     }
 
 
+def _ensure_environments_table(conn):
+    """幂等建环境表：server init_db 与各环境接口都可调它，老库
+    第一次跑到这里时自动补表，不丢数据。类型写法与 followups
+    表一致（DOUBLE PRECISION 在 SQLite 与 Postgres 下都可用）。
+
+    环境 = 项目仓库的一个命名基准检出：repo_url 是创建时从
+    项目快照来的（之后项目改 repo_url 不影响已建环境），branch
+    为空表示默认分支，executor 为空表示不限执行器（任务创建时
+    若指定执行器须与它一致，否则继承它）。状态机：creating
+    （已记录、等 worker 准备检出）-> ready / failed；archive
+    后新任务不能再选它，旧任务结果不受影响。检出目录本身不
+    存绝对路径（worker 按 repos_dir / 租户目录推导，见 worker
+    的 _env_host_dir），避免 server 与 worker 机器路径不一致。
+    """
+    conn.execute(
+        "CREATE TABLE IF NOT EXISTS environments("
+        "id TEXT PRIMARY KEY, "
+        "project_id TEXT, "
+        "name TEXT, "
+        "branch TEXT, "
+        "executor TEXT, "
+        "status TEXT, "
+        "error_text TEXT, "
+        "repo_url TEXT, "
+        "created_by TEXT, "
+        "created_at DOUBLE PRECISION, "
+        "updated_at DOUBLE PRECISION)"
+    )
+    conn.execute(
+        "CREATE INDEX IF NOT EXISTS idx_env_project"
+        " ON environments(project_id, status)")
+    # 同项目未归档环境名唯一：用部分唯一索引（SQLite 与 Postgres
+    # 都支持 WHERE 子句），归档过的名字可重用——全表 UNIQUE
+    # 会挡住重用，与 _create_environment 的 409 规则矛盾。
+    conn.execute(
+        "CREATE UNIQUE INDEX IF NOT EXISTS idx_env_name"
+        " ON environments(project_id, name)"
+        " WHERE status != 'archived'")
+
+
+_ENV_BRANCH_RE = re.compile(r"^[A-Za-z0-9_.\-/]+$")
+
+
+def _validate_env_branch(branch):
+    """环境分支名校验：空/None 表示默认分支；非空时只允许
+    字母数字与 _ . - /，且不能以 - 开头、不能含 ..、不能以
+    .lock 结尾、长度不超过 200——git ref 名的保守子集，防
+    分支名被拼进 shell 或越出目录。返回规范化后的分支或 None，
+    非法时抛 ValueError（调用方转 400）。"""
+    if branch is None:
+        return None
+    if not isinstance(branch, str):
+        raise ValueError("branch must be a string")
+    branch = branch.strip()
+    if not branch:
+        return None
+    if len(branch) > 200 or not _ENV_BRANCH_RE.match(branch) \
+            or branch.startswith("-") or ".." in branch \
+            or branch.endswith(".lock"):
+        raise ValueError("branch is not a valid git branch name")
+    return branch
+
+
+def _environment_to_dict(row):
+    return {
+        "id": row["id"],
+        "project_id": row["project_id"],
+        "name": row["name"],
+        "branch": _row_get(row, "branch"),
+        "executor": _row_get(row, "executor"),
+        "status": row["status"],
+        "error_text": _row_get(row, "error_text"),
+        "repo_url": _row_get(row, "repo_url"),
+        "created_by": row["created_by"],
+        "created_at": row["created_at"],
+        "updated_at": row["updated_at"],
+    }
+
+
 def _is_postgres_backend():
     backend = os.environ.get("CLOUD_DB_BACKEND") or "sqlite"
     return str(backend).strip().lower() in ("postgres", "postgresql")
@@ -530,6 +613,7 @@ def init_db(db_path):
             migrate_tasks_table(conn)
             migrate_projects_table(conn)
             _ensure_followups_table(conn)
+            _ensure_environments_table(conn)
             conn.commit()
             return
         conn.execute(
@@ -559,6 +643,7 @@ def init_db(db_path):
         migrate_tasks_table(conn)
         migrate_projects_table(conn)
         _ensure_followups_table(conn)
+        _ensure_environments_table(conn)
         # worker 每轮 poll 都按 status 取任务（见 load_tasks 的
         # WHERE status IN (...)），无索引时是全表扫描；任务表随
         # 时间变大后索引是必须的。IF NOT EXISTS 保证老库幂等。
@@ -630,6 +715,7 @@ def _task_to_dict(row):
         "approval": _row_get(row, "approval"),
         "plan": _row_get(row, "plan") or 0,
         "forked_from": _row_get(row, "forked_from"),
+        "environment_id": _row_get(row, "environment_id"),
     }
 
 
@@ -927,6 +1013,18 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             self._relay_models(user_id)
         elif path.startswith("/projects/"):
             self._dispatch_project(user_id, method, path, query)
+        elif path == "/environments" and method == "POST":
+            self._create_environment(user_id)
+        elif path == "/environments" and method == "GET":
+            self._list_environments(user_id, query)
+        elif path.startswith("/environments/"):
+            rest = path[len("/environments/"):]
+            if "/" not in rest and method == "GET":
+                self._get_environment(user_id, rest)
+            elif rest.endswith("/archive") and method == "POST":
+                self._archive_environment(user_id, rest[: -len("/archive")])
+            else:
+                self._send_json(404, {"error": "not found"})
         elif path == "/tasks" and method == "POST":
             self._create_task(user_id)
         elif path == "/tasks" and method == "GET":
@@ -969,6 +1067,190 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         else:
             self._send_json(404, {"error": "not found"})
 
+    # -- 环境 ----------------------------------------------------------
+    def _fetch_environment(self, env_id):
+        conn = self._db()
+        try:
+            return conn.execute(
+                "SELECT * FROM environments WHERE id = ?", (env_id,)
+            ).fetchone()
+        finally:
+            conn.close()
+
+    def _create_environment(self, user_id):
+        """建环境：只记录意图（status=creating），检出由 worker
+        在下一轮准备（clone + checkout 分支），准备好后 status
+        变 ready，失败变 failed 并带 error_text——创建本身是同步
+        返回的记录，准备是异步的，与任务 queued->running 同款语义。
+        同项目内未归档环境名唯一（409），避免选环境时分不清；
+        归档过的名字可重用。项目没有 repo_url 时 400：环境就是
+        仓库的基准检出，没有仓库就没有环境。"""
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        project_id = body.get("project_id")
+        if not project_id or not isinstance(project_id, str):
+            self._send_json(400, {"error": "project_id required"})
+            return
+        name = body.get("name")
+        if not name or not isinstance(name, str) or not name.strip():
+            self._send_json(400, {"error": "name required"})
+            return
+        name = name.strip()
+        if len(name) > 100:
+            self._send_json(400, {"error": "name too long"})
+            return
+        executor = body.get("executor")
+        if executor is not None and executor not in ("local", "docker"):
+            self._send_json(400, {
+                "error": "executor must be local or docker"})
+            return
+        try:
+            branch = _validate_env_branch(body.get("branch"))
+        except ValueError as exc:
+            self._send_json(400, {"error": str(exc)})
+            return
+        err = self._check_project(user_id, project_id, "create_task")
+        if err:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        now = time.time()
+        env_id = uuid.uuid4().hex
+        conn = self._db()
+        try:
+            proj = conn.execute(
+                "SELECT repo_url FROM projects WHERE id = ?",
+                (project_id,)).fetchone()
+            if proj is None:
+                self._send_json(404, {"error": "project not found"})
+                return
+            repo_url = _row_get(proj, "repo_url")
+            if not repo_url:
+                self._send_json(400, {
+                    "error": "project has no repo_url, "
+                    "environment needs a repo"})
+                return
+            existing = conn.execute(
+                "SELECT id FROM environments WHERE project_id = ?"
+                " AND name = ? AND status != 'archived'",
+                (project_id, name)).fetchone()
+            if existing:
+                self._send_json(409, {
+                    "error": "environment name already exists"
+                    " in this project"})
+                return
+            try:
+                conn.execute(
+                    "INSERT INTO environments(id, project_id, name,"
+                    " branch, executor, status, error_text, repo_url,"
+                    " created_by, created_at, updated_at)"
+                    " VALUES(?, ?, ?, ?, ?, 'creating', NULL, ?, ?,"
+                    " ?, ?)",
+                    (env_id, project_id, name, branch, executor,
+                     repo_url, user_id, now, now))
+                conn.commit()
+            except (sqlite3.IntegrityError, _PgIntegrityError):
+                self._send_json(409, {
+                    "error": "environment name already exists"
+                    " in this project"})
+                return
+            row = conn.execute(
+                "SELECT * FROM environments WHERE id = ?", (env_id,)
+            ).fetchone()
+        finally:
+            conn.close()
+        with _auth_store(self.server.db_path) as store:
+            store.audit(user_id, project_id, "environment_created",
+                        env_id)
+        self._send_json(201, _environment_to_dict(row))
+
+    def _list_environments(self, user_id, query):
+        project_ids = query.get("project_id")
+        project_id = project_ids[0] if project_ids else None
+        if not project_id:
+            self._send_json(400, {"error": "project_id required"})
+            return
+        err = self._check_project(user_id, project_id, "read")
+        if err:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        include_archived = query.get("include_archived") == ["1"]
+        conn = self._db()
+        try:
+            if include_archived:
+                rows = conn.execute(
+                    "SELECT * FROM environments WHERE project_id = ?"
+                    " ORDER BY created_at, id",
+                    (project_id,)).fetchall()
+            else:
+                rows = conn.execute(
+                    "SELECT * FROM environments WHERE project_id = ?"
+                    " AND status != 'archived'"
+                    " ORDER BY created_at, id",
+                    (project_id,)).fetchall()
+        finally:
+            conn.close()
+        self._send_json(200, {
+            "environments": [_environment_to_dict(r) for r in rows]})
+
+    def _get_environment(self, user_id, env_id):
+        row = self._fetch_environment(env_id)
+        if row is None:
+            self._send_json(404, {"error": "environment not found"})
+            return
+        env = _environment_to_dict(row)
+        err = self._check_project(user_id, env["project_id"], "read")
+        if err:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        self._send_json(200, env)
+
+    def _archive_environment(self, user_id, env_id):
+        """归档环境：已归档时幂等返回 200；有 queued/running
+        任务正在用它时 409，绝不让进行中的任务失去基准检出。
+        归档只改状态，不删检出目录（已完成任务的 worktree 独立，
+        检出留着可查；磁盘清理另行手动做，不在归档里顺手删）。"""
+        row = self._fetch_environment(env_id)
+        if row is None:
+            self._send_json(404, {"error": "environment not found"})
+            return
+        env = _environment_to_dict(row)
+        err = self._check_project(
+            user_id, env["project_id"], "create_task")
+        if err:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        if env["status"] == "archived":
+            self._send_json(200, env)
+            return
+        now = time.time()
+        conn = self._db()
+        try:
+            using = conn.execute(
+                "SELECT id FROM tasks WHERE environment_id = ?"
+                " AND status IN ('queued', 'running') LIMIT 1",
+                (env_id,)).fetchone()
+            if using:
+                self._send_json(409, {
+                    "error": "environment is in use by a queued or"
+                    " running task"})
+                return
+            conn.execute(
+                "UPDATE environments SET status = 'archived',"
+                " updated_at = ? WHERE id = ?",
+                (now, env_id))
+            conn.commit()
+            row = conn.execute(
+                "SELECT * FROM environments WHERE id = ?", (env_id,)
+            ).fetchone()
+        finally:
+            conn.close()
+        with _auth_store(self.server.db_path) as store:
+            store.audit(user_id, env["project_id"],
+                        "environment_archived", env_id)
+        self._send_json(200, _environment_to_dict(row))
+
     # -- 项目 ----------------------------------------------------------
     def _create_project(self, user_id):
         body, err = self._read_json()
@@ -1599,6 +1881,46 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             if not store.check_permission(user_id, project_id, "create_task"):
                 self._send_json(403, {"error": "forbidden"})
                 return
+        # 环境绑定：environment_id 非空时必须是本项目、已 ready
+        # 的环境；creating/failed/archived 一律拒绝并把状态写进
+        # 报错（creating 时用户就知道该等 worker 准备好再建任务，
+        # 而不是任务排队后才在 worker 侧失败）。执行器规则：
+        # 环境 executor 非空且任务 executor 已指定时必须一致；
+        # 任务 executor 未指定时继承环境 executor（认领过滤靠它，
+        # 否则 local 任务可能被 docker worker 认领、反之亦然）；
+        # cloud 执行器没有环境检出路径，一律拒绝。
+        environment_id = body.get("environment_id")
+        if environment_id is not None:
+            if not isinstance(environment_id, str) \
+                    or not environment_id.strip():
+                self._send_json(400, {
+                    "error": "environment_id must be a non-empty string"})
+                return
+            environment_id = environment_id.strip()
+            env_row = self._fetch_environment(environment_id)
+            if env_row is None:
+                self._send_json(404, {"error": "environment not found"})
+                return
+            env = _environment_to_dict(env_row)
+            if env["project_id"] != project_id:
+                self._send_json(400, {
+                    "error": "environment belongs to another project"})
+                return
+            if env["status"] != "ready":
+                self._send_json(400, {
+                    "error": "environment not ready: %s" % env["status"]})
+                return
+            if executor == "cloud":
+                self._send_json(400, {
+                    "error": "environment does not support cloud executor"})
+                return
+            if env["executor"] and executor \
+                    and env["executor"] != executor:
+                self._send_json(400, {
+                    "error": "executor does not match environment executor"})
+                return
+            if executor is None and env["executor"]:
+                executor = env["executor"]
         now = time.time()
         conn = self._db()
         try:
@@ -1618,13 +1940,15 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                     " idempotency_key, depends_on, priority, created_by,"
                     " created_at, updated_at, last_heartbeat, executor,"
                     " goal, verify_command, goal_max_rounds,"
-                    " model, effort, approval, plan, goal_mode)"
+                    " model, effort, approval, plan, goal_mode,"
+                    " environment_id)"
                     " VALUES(?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, NULL, ?,"
-                    " ?, ?, ?, ?, ?, ?, ?, ?)",
+                    " ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                     (task_id, project_id, prompt, idempotency_key,
                      json.dumps(depends_on), priority, user_id, now, now,
                      executor, goal, verify_command, goal_max_rounds,
-                     model, effort, approval, plan, goal_mode),
+                     model, effort, approval, plan, goal_mode,
+                     environment_id),
                 )
                 conn.commit()
             except (sqlite3.IntegrityError, _PgIntegrityError):

cloud/tests/test_task_environments.py

@@ -0,0 +1,499 @@
+"""云端环境单测：选、建、管 + 任务绑定环境。
+
+不跑真实模型回合、不联网：server 侧走本地临时端口 + 临时
+SQLite；worker 侧注入假 git runner（clone/fetch/checkout 只
+操作临时目录，不调真实 git）；console 侧只查 HTML 标记。
+
+链路覆盖：建环境（creating）-> worker 准备（ready/failed）
+-> 建任务选环境（执行器继承/冲突拒绝）-> 任务派发指到环境
+检出 -> 归档（进行中任务在用时拒绝、幂等）。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import time
+import unittest
+import uuid
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.sandbox import DockerExecutor
+    from cloud.server import create_server, init_db
+    from cloud.worker import Worker
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.sandbox import DockerExecutor
+    from cloud.server import create_server, init_db
+    from cloud.worker import Worker
+
+
+def _req(port, method, path, body=None, key=None):
+    conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10)
+    hdrs = {}
+    payload = None
+    if body is not None:
+        payload = json.dumps(body).encode("utf-8")
+        hdrs["Content-Type"] = "application/json"
+    if key is not None:
+        hdrs["X-API-Key"] = key
+    try:
+        conn.request(method, path, body=payload, headers=hdrs)
+        resp = conn.getresponse()
+        data = resp.read()
+    finally:
+        conn.close()
+    try:
+        parsed = json.loads(data.decode("utf-8")) if data else None
+    except ValueError:
+        parsed = None
+    return resp.status, parsed
+
+
+def _set_env_status(db_path, env_id, status):
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute("UPDATE environments SET status = ? WHERE id = ?",
+                     (status, env_id))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _set_task_status(db_path, task_id, status):
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute("UPDATE tasks SET status = ? WHERE id = ?",
+                     (status, task_id))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+class ServerEnvironmentsTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.other_id, self.other_key = self.auth.create_user("other")
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, self.project = _req(
+            self.port, "POST", "/projects",
+            {"name": "demo", "repo_url": "https://github.com/octocat/Hello-World.git"},
+            self.owner_key)
+        self.assertEqual(status, 201, self.project)
+        status, self.plain_project = _req(
+            self.port, "POST", "/projects", {"name": "plain"},
+            self.owner_key)
+        self.assertEqual(status, 201, self.plain_project)
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _create_env(self, project_id=None, name="dev", **extra):
+        body = {"project_id": project_id or self.project["id"],
+                "name": name}
+        body.update(extra)
+        return _req(self.port, "POST", "/environments", body,
+                    self.owner_key)
+
+    def _ready_env(self, name="dev", **extra):
+        status, env = self._create_env(name=name, **extra)
+        self.assertEqual(status, 201, env)
+        _set_env_status(self.db_path, env["id"], "ready")
+        return env
+
+    def test_create_environment_roundtrip(self):
+        status, env = self._create_env(
+            name="dev", branch="main", executor="local")
+        self.assertEqual(status, 201, env)
+        self.assertEqual(env["status"], "creating")
+        self.assertEqual(env["project_id"], self.project["id"])
+        self.assertEqual(env["branch"], "main")
+        self.assertEqual(env["executor"], "local")
+        self.assertEqual(env["repo_url"],
+                         "https://github.com/octocat/Hello-World.git")
+        status, got = _req(
+            self.port, "GET", "/environments/" + env["id"],
+            key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(got["id"], env["id"])
+
+    def test_create_environment_validation(self):
+        # 项目没有 repo_url：环境无从检出。
+        status, _ = self._create_env(
+            project_id=self.plain_project["id"], name="x")
+        self.assertEqual(status, 400)
+        # 名字、分支、执行器非法。
+        status, _ = self._create_env(name="  ")
+        self.assertEqual(status, 400)
+        status, _ = self._create_env(name="x", branch="../evil")
+        self.assertEqual(status, 400)
+        status, _ = self._create_env(name="x", branch="-bad")
+        self.assertEqual(status, 400)
+        status, _ = self._create_env(name="x", executor="cloud")
+        self.assertEqual(status, 400)
+        # 非成员、不存在的环境。
+        status, _ = _req(
+            self.port, "POST", "/environments",
+            {"project_id": self.project["id"], "name": "x"},
+            self.other_key)
+        self.assertEqual(status, 403)
+        status, _ = _req(self.port, "GET", "/environments/nope",
+                         key=self.owner_key)
+        self.assertEqual(status, 404)
+
+    def test_environment_name_unique_per_project(self):
+        status, _ = self._create_env(name="dev")
+        self.assertEqual(status, 201)
+        status, _ = self._create_env(name="dev")
+        self.assertEqual(status, 409)
+        # 另一项目同名可以；归档后原项目同名可重用。
+        status, env2 = self._create_env(
+            project_id=self.plain_project["id"], name="dev")
+        # plain 项目无 repo_url，上面应 400——换一个带仓库的项目测同名。
+        self.assertEqual(status, 400)
+        status, other = _req(
+            self.port, "POST", "/projects",
+            {"name": "demo2",
+             "repo_url": "https://github.com/octocat/Hello-World.git"},
+            self.owner_key)
+        self.assertEqual(status, 201)
+        status, _ = self._create_env(project_id=other["id"], name="dev")
+        self.assertEqual(status, 201)
+
+    def test_list_environments_and_archive(self):
+        env = self._ready_env(name="dev")
+        status, listing = _req(
+            self.port, "GET",
+            "/environments?project_id=" + self.project["id"],
+            key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(len(listing["environments"]), 1)
+        # 缺 project_id 400；非成员 403。
+        status, _ = _req(self.port, "GET", "/environments",
+                         key=self.owner_key)
+        self.assertEqual(status, 400)
+        status, _ = _req(
+            self.port, "GET",
+            "/environments?project_id=" + self.project["id"],
+            key=self.other_key)
+        self.assertEqual(status, 403)
+        # 归档后默认列表排除，include_archived=1 才出现；再归档幂等。
+        status, archived = _req(
+            self.port, "POST", "/environments/%s/archive" % env["id"],
+            {}, self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(archived["status"], "archived")
+        status, listing = _req(
+            self.port, "GET",
+            "/environments?project_id=" + self.project["id"],
+            key=self.owner_key)
+        self.assertEqual(listing["environments"], [])
+        status, listing = _req(
+            self.port, "GET",
+            "/environments?project_id=" + self.project["id"]
+            + "&include_archived=1",
+            key=self.owner_key)
+        self.assertEqual(len(listing["environments"]), 1)
+        status, again = _req(
+            self.port, "POST", "/environments/%s/archive" % env["id"],
+            {}, self.owner_key)
+        self.assertEqual(status, 200)
+
+    def test_archive_blocked_when_task_using_it(self):
+        env = self._ready_env(name="dev")
+        status, task = _req(
+            self.port, "POST", "/tasks",
+            {"project_id": self.project["id"], "prompt": "p",
+             "environment_id": env["id"]},
+            self.owner_key)
+        self.assertEqual(status, 201, task)
+        status, _ = _req(
+            self.port, "POST", "/environments/%s/archive" % env["id"],
+            {}, self.owner_key)
+        self.assertEqual(status, 409)
+        _set_task_status(self.db_path, task["id"], "completed")
+        status, archived = _req(
+            self.port, "POST", "/environments/%s/archive" % env["id"],
+            {}, self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(archived["status"], "archived")
+
+    def test_task_with_environment_rules(self):
+        # creating 的环境不能建任务。
+        status, env = self._create_env(name="dev", executor="docker")
+        self.assertEqual(status, 201)
+        status, _ = _req(
+            self.port, "POST", "/tasks",
+            {"project_id": self.project["id"], "prompt": "p",
+             "environment_id": env["id"]},
+            self.owner_key)
+        self.assertEqual(status, 400)
+        # ready 后：执行器继承、冲突拒绝、cloud 拒绝、跨项目拒绝。
+        _set_env_status(self.db_path, env["id"], "ready")
+        status, task = _req(
+            self.port, "POST", "/tasks",
+            {"project_id": self.project["id"], "prompt": "p",
+             "environment_id": env["id"]},
+            self.owner_key)
+        self.assertEqual(status, 201, task)
+        self.assertEqual(task["executor"], "docker")
+        self.assertEqual(task["environment_id"], env["id"])
+        status, _ = _req(
+            self.port, "POST", "/tasks",
+            {"project_id": self.project["id"], "prompt": "p",
+             "environment_id": env["id"], "executor": "local"},
+            self.owner_key)
+        self.assertEqual(status, 400)
+        status, _ = _req(
+            self.port, "POST", "/tasks",
+            {"project_id": self.project["id"], "prompt": "p",
+             "environment_id": env["id"], "executor": "cloud"},
+            self.owner_key)
+        self.assertEqual(status, 400)
+        status, _ = _req(
+            self.port, "POST", "/tasks",
+            {"project_id": self.plain_project["id"], "prompt": "p",
+             "environment_id": env["id"]},
+            self.owner_key)
+        self.assertEqual(status, 400)
+        status, _ = _req(
+            self.port, "POST", "/tasks",
+            {"project_id": self.project["id"], "prompt": "p",
+             "environment_id": "nope"},
+            self.owner_key)
+        self.assertEqual(status, 404)
+        # failed 的环境也不能建任务。
+        _set_env_status(self.db_path, env["id"], "failed")
+        status, _ = _req(
+            self.port, "POST", "/tasks",
+            {"project_id": self.project["id"], "prompt": "p",
+             "environment_id": env["id"]},
+            self.owner_key)
+        self.assertEqual(status, 400)
+
+    def test_console_has_environment_ui(self):
+        path = Path(__file__).resolve().parents[1] / "console.html"
+        html = path.read_text(encoding="utf-8")
+        for marker in ("envSelect", "loadEnvironments",
+                       "createEnvironment", "refreshEnvironments",
+                       "newEnvName", "newEnvBranch", "newEnvExecutor",
+                       "/environments?project_id="):
+            self.assertIn(marker, html)
+
+
+class FakeGit:
+    """假 git：clone 在临时目录建 .git 并记 origin；fetch 恒成功；
+    checkout 只接受白名单分支，其余报分支不存在。"""
+
+    def __init__(self, allowed_branches=("main", "feature/x")):
+        self.allowed = set(allowed_branches)
+        self.origins = {}
+        self.commands = []
+
+    def __call__(self, cmd):
+        self.commands.append(list(cmd))
+        if cmd[1] == "clone":
+            target = cmd[3]
+            os.makedirs(os.path.join(target, ".git"), exist_ok=True)
+            self.origins[target] = cmd[2]
+            return SimpleNamespace(returncode=0, stdout="", stderr="")
+        if cmd[1] == "-C":
+            target = cmd[2]
+            if cmd[3] == "remote":
+                return SimpleNamespace(
+                    returncode=0, stdout=self.origins.get(target, ""),
+                    stderr="")
+            if cmd[3] == "fetch":
+                return SimpleNamespace(returncode=0, stdout="", stderr="")
+            if cmd[3] == "checkout":
+                branch = cmd[4]
+                if branch in self.allowed:
+                    return SimpleNamespace(returncode=0, stdout="",
+                                           stderr="")
+                return SimpleNamespace(
+                    returncode=1, stdout="",
+                    stderr="pathspec '%s' did not match" % branch)
+        return SimpleNamespace(returncode=1, stdout="",
+                               stderr="unexpected git command")
+
+
+def _insert_env(db_path, env_id="env1", project_id="p1",
+                status="creating", branch=None, executor=None,
+                name="dev",
+                repo_url="https://github.com/octocat/Hello-World.git"):
+    now = time.time()
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute(
+            "INSERT INTO environments(id, project_id, name, branch,"
+            " executor, status, error_text, repo_url, created_by,"
+            " created_at, updated_at) VALUES(?, ?, ?, ?, ?, ?,"
+            " NULL, ?, 'tester', ?, ?)",
+            (env_id, project_id, name, branch, executor, status,
+             repo_url, now, now))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _env_row(db_path, env_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        row = conn.execute(
+            "SELECT * FROM environments WHERE id = ?",
+            (env_id,)).fetchone()
+        cols = [d[0] for d in conn.execute(
+            "SELECT * FROM environments LIMIT 0").description]
+        return dict(zip(cols, row)) if row else None
+    finally:
+        conn.close()
+
+
+class WorkerEnvironmentsTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        self.repos_dir = os.path.join(self.tmp.name, "repos")
+        self.env = {"XCM_API_KEY": "test-placeholder"}
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _worker(self, git=None, **kwargs):
+        kwargs.setdefault("env", self.env)
+        kwargs.setdefault("repos_dir", self.repos_dir)
+        kwargs.setdefault("repos_runner", git)
+        return Worker(self.db_path, "/fake/host.py", **kwargs)
+
+    def test_prepare_environment_ready(self):
+        _insert_env(self.db_path, branch="feature/x")
+        worker = self._worker(git=FakeGit())
+        results = worker._prepare_pending_environments()
+        self.assertEqual(results, [("env1", "ready", None)])
+        row = _env_row(self.db_path, "env1")
+        self.assertEqual(row["status"], "ready")
+        target = os.path.join(self.repos_dir, "envs", "env1")
+        self.assertTrue(os.path.isdir(os.path.join(target, ".git")))
+
+    def test_prepare_environment_failed_branch(self):
+        _insert_env(self.db_path, branch="no-such-branch")
+        worker = self._worker(git=FakeGit())
+        results = worker._prepare_pending_environments()
+        self.assertEqual(results[0][1], "failed")
+        self.assertIn("checkout", results[0][2])
+        self.assertEqual(_env_row(self.db_path, "env1")["status"],
+                         "failed")
+
+    def test_prepare_environment_archived_during_prepare(self):
+        # 准备期间被归档：CAS 改不到行，状态保持 archived。
+        _insert_env(self.db_path)
+        worker = self._worker(git=FakeGit())
+        orig = worker._prepare_environment
+
+        def _prepare_and_archive(env):
+            err = orig(env)
+            _set_env_status(self.db_path, env["id"], "archived")
+            return err
+
+        worker._prepare_environment = _prepare_and_archive
+        results = worker._prepare_pending_environments()
+        self.assertEqual(results[0][1], "archived")
+        self.assertEqual(_env_row(self.db_path, "env1")["status"],
+                         "archived")
+
+    def test_apply_environment_repo_local(self):
+        _insert_env(self.db_path, status="ready", executor="local")
+        worker = self._worker()
+        task = {"id": "T", "project_id": "p1", "environment_id": "env1",
+                "executor": "local"}
+        self.assertIsNone(worker._apply_project_repo(task))
+        self.assertEqual(
+            task["repo"],
+            os.path.join(self.repos_dir, "envs", "env1"))
+
+    def test_apply_environment_repo_tenant_docker(self):
+        _insert_env(self.db_path, status="ready", executor="docker")
+        worker = self._worker(
+            executor=DockerExecutor(config_path="/fake/cfg.json"),
+            tenant_isolation=True,
+            tenants_root=os.path.join(self.tmp.name, "tenants"))
+        task = {"id": "T", "project_id": "p1", "environment_id": "env1",
+                "executor": "docker"}
+        self.assertIsNone(worker._apply_project_repo(task))
+        self.assertEqual(task["repo"],
+                         "/var/lib/codex-app-server/envs/env1")
+
+    def test_apply_environment_repo_rejections(self):
+        worker = self._worker()
+        # 环境不存在。
+        task = {"id": "T", "project_id": "p1", "environment_id": "nope"}
+        self.assertIn("not found", worker._apply_project_repo(task))
+        # 未 ready、跨项目、执行器冲突（同项目多环境须不同名，
+        # 部分唯一索引只拦同项目同名未归档行）。
+        _insert_env(self.db_path, env_id="e2", name="e2",
+                    status="creating")
+        task = {"id": "T", "project_id": "p1", "environment_id": "e2"}
+        self.assertIn("not ready", worker._apply_project_repo(task))
+        _insert_env(self.db_path, env_id="e3", project_id="p2",
+                    name="e3", status="ready")
+        task = {"id": "T", "project_id": "p1", "environment_id": "e3"}
+        self.assertIn("another project",
+                      worker._apply_project_repo(task))
+        _insert_env(self.db_path, env_id="e4", name="e4",
+                    status="ready", executor="docker")
+        task = {"id": "T", "project_id": "p1", "environment_id": "e4",
+                "executor": "local"}
+        self.assertIn("does not match",
+                      worker._apply_project_repo(task))
+
+    def test_tenant_mounts_include_envs(self):
+        worker = self._worker(
+            executor=DockerExecutor(config_path="/fake/cfg.json"),
+            tenant_isolation=True,
+            tenants_root=os.path.join(self.tmp.name, "tenants"))
+        # 先播种租户目录（codex-home 模板缺失会报错，这里只测
+        # envs 挂载条件：手动建好租户目录结构再调 _tenant_mounts
+        # 的 envs 分支——目录不存在时不挂载，存在时挂载）。
+        base = os.path.join(self.tmp.name, "tenants", "p1")
+        os.makedirs(os.path.join(base, "work"), exist_ok=True)
+        os.makedirs(os.path.join(base, "codex-home"), exist_ok=True)
+        with open(os.path.join(base, "codex-home", "config.toml"),
+                  "w") as f:
+            f.write("")
+        os.makedirs(os.path.join(base, "test-repo"), exist_ok=True)
+        with open(os.path.join(base, "test-repo", "sample.py"),
+                  "w") as f:
+            f.write("")
+        mounts = worker._tenant_mounts("p1")
+        self.assertFalse(any("envs" in m for m in mounts))
+        os.makedirs(os.path.join(base, "envs", "env1"), exist_ok=True)
+        mounts = worker._tenant_mounts("p1")
+        self.assertTrue(any(
+            m.endswith(":/var/lib/codex-app-server/envs:rw")
+            for m in mounts))
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/worker.py

@@ -138,7 +138,7 @@ from cloud.models import ModelRouter, is_fallbackable, record_fallback
 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
-from cloud.repos import RepoError, resolve_repo
+from cloud.repos import RepoError, checkout_dir, resolve_repo
 from cloud.sandbox import CloudExecutor, DockerExecutor, LocalExecutor
 from cloud.sandbox import _default_runner as _sandbox_default_runner
 from cloud.workers import heartbeat_worker, register_worker
@@ -243,6 +243,8 @@ def _task_from_row(row):
         "approval": _col("approval"),
         "plan": _col("plan"),
         "forked_from": _col("forked_from"),
+        # 环境：老行缺列时 None = 不指定环境，走项目仓库路径。
+        "environment_id": _col("environment_id"),
     }
 
 
@@ -470,7 +472,7 @@ class Worker:
                 "SELECT id, project_id, prompt, status, idempotency_key,"
                 " depends_on, priority, created_by, created_at, updated_at,"
                 " last_heartbeat, executor, goal, verify_command,"
-                " goal_max_rounds, goal_mode FROM tasks"
+                " goal_max_rounds, goal_mode, environment_id FROM tasks"
                 " WHERE status IN ('queued', 'completed', 'merged')"
                 " ORDER BY created_at, id").fetchall()
         finally:
@@ -571,7 +573,8 @@ class Worker:
                 "effort": "TEXT",
                 "approval": "TEXT",
                 "plan": "INTEGER",
-                "forked_from": "TEXT"}.items():
+                "forked_from": "TEXT",
+                "environment_id": "TEXT"}.items():
             if name not in cols:
                 try:
                     conn.execute(
@@ -1087,12 +1090,231 @@ class Worker:
                 "%s:/var/lib/codex-app-server/repos/%s:rw"
                 % (dirs["repo_checkout"],
                    self._tenant_safe_name(project_id)))
+        # 环境检出：整目录挂载（按项目分，与 repo_checkout 同款
+        # 隔离语义）。环境任务派发前 worker 已准备好检出，目录
+        # 存在时才挂载；不存在说明该项目暂无环境，不硬挂载。
+        # base = work 的父目录（即租户项目根目录）。
+        envs_dir = os.path.join(os.path.dirname(dirs["work"]), "envs")
+        if os.path.isdir(envs_dir):
+            mounts.append(
+                "%s:/var/lib/codex-app-server/envs:rw" % envs_dir)
         return mounts
 
     def _tenant_network(self, project_id):
         """某项目的 Docker 网络名（容器跨项目网络隔离用）。"""
         return "codex-" + self._tenant_safe_name(project_id)
 
+    # -- 环境 ----------------------------------------------------------
+    def _env_host_dir(self, env):
+        """环境的宿主检出目录（worker 准备检出时写这里）。
+
+        非租户：repos_dir/envs/<env_id>（与项目检出同根，Docker
+        非租户挂载 repos 根目录时它自然在容器可见路径下）。
+        租户 + Docker：tenants_root/<项目>/envs/<env_id>，由
+        _tenant_mounts 挂到容器 /var/lib/codex-app-server/envs。
+        env_id 是 server 生成的 hex，仍按 checkout_dir 同款规则
+        过滤安全字符，防路径穿越。"""
+        if self.tenant_isolation and isinstance(
+                self.executor, DockerExecutor):
+            safe_proj = self._tenant_safe_name(env["project_id"])
+            return checkout_dir(
+                os.path.join(self.tenants_root, safe_proj, "envs"),
+                env["id"])
+        return checkout_dir(
+            os.path.join(self.repos_dir, "envs"), env["id"])
+
+    def _env_repo_path(self, env):
+        """任务派发时传给执行器的 repo 路径：租户 Docker 用容器
+        内路径，其余用宿主路径——与 _apply_project_repo 的项目
+        检出路径约定完全一致，执行器不用为环境另起分支。"""
+        if self.tenant_isolation and isinstance(
+                self.executor, DockerExecutor):
+            return "/var/lib/codex-app-server/envs/%s" % env["id"]
+        return self._env_host_dir(env)
+
+    def _fetch_environment(self, env_id):
+        """读环境行；表不存在（老库未跑 server init_db）或查不到
+        时返回 None，调用方按“环境不存在”处理，绝不炸 worker。"""
+        try:
+            conn = _db_connect(self.db_path)
+            try:
+                row = conn.execute(
+                    "SELECT * FROM environments WHERE id = ?",
+                    (env_id,)).fetchone()
+            finally:
+                conn.close()
+        except Exception:
+            return None
+        if row is None:
+            return None
+
+        def _get(name):
+            try:
+                return row[name]
+            except (KeyError, IndexError, TypeError):
+                return None
+
+        try:
+            return {
+                "id": row["id"],
+                "project_id": row["project_id"],
+                "name": row["name"],
+                "branch": _get("branch"),
+                "executor": _get("executor"),
+                "status": row["status"],
+                "error_text": _get("error_text"),
+                "repo_url": _get("repo_url"),
+            }
+        except (KeyError, IndexError, TypeError):
+            return None
+
+    def _run_git(self, cmd):
+        """跑一条 git 命令：repos_runner 可注入（单测用假 runner，
+        接受 cmd 或 (cmd, timeout) 两种签名，与 repos._run 同款
+        兼容）；缺省走真实 subprocess。返回带 returncode/stdout/
+        stderr 的结果对象。"""
+        if self.repos_runner is not None:
+            try:
+                return self.repos_runner(cmd)
+            except TypeError:
+                return self.repos_runner(cmd, timeout=None)
+        return subprocess.run(cmd, capture_output=True, text=True)
+
+    def _prepare_environment(self, env):
+        """准备一个环境的检出，返回错误文本或 None。
+
+        幂等：目录不存在时 clone，存在时先校验它是 git 仓库且
+        origin 与环境 repo_url 一致（不一致绝不强制改 remote，
+        直接失败），再 fetch；分支非空时 checkout 该分支（不存
+        在则失败，绝不自动建分支——分支名是用户选的，不应由
+        准备过程发明）。检出本身不做 reset/clean：任务 worktree
+        由 host.py 从检出 HEAD 新建，检出保持干净即可。"""
+        repo_url = env.get("repo_url")
+        if not repo_url:
+            return "environment has no repo_url"
+        target = self._env_host_dir(env)
+        if not os.path.exists(target):
+            os.makedirs(os.path.dirname(target), exist_ok=True)
+            result = self._run_git(
+                ["git", "clone", repo_url, target])
+            if getattr(result, "returncode", 1) != 0:
+                err = (getattr(result, "stderr", "") or "").strip()
+                return "git clone 失败: %s" % (err or repo_url)
+        else:
+            git_path = os.path.join(target, ".git")
+            if not (os.path.isdir(git_path)
+                    or os.path.isfile(git_path)):
+                return "检出目录已存在但不是 git 仓库，拒绝覆盖: %s" \
+                    % target
+            origin = self._run_git(
+                ["git", "-C", target, "remote", "get-url", "origin"])
+            if getattr(origin, "returncode", 1) != 0:
+                return "检出目录 %s 取不到 remote origin，拒绝准备" \
+                    % target
+            if (getattr(origin, "stdout", "") or "").strip() != repo_url:
+                return "检出目录 origin 与环境 repo_url 不一致，拒绝准备"
+            fetched = self._run_git(
+                ["git", "-C", target, "fetch", "origin"])
+            if getattr(fetched, "returncode", 1) != 0:
+                err = (getattr(fetched, "stderr", "") or "").strip()
+                return "git fetch origin 失败: %s" % (err or target)
+        branch = env.get("branch")
+        if branch:
+            checked = self._run_git(
+                ["git", "-C", target, "checkout", branch])
+            if getattr(checked, "returncode", 1) != 0:
+                err = (getattr(checked, "stderr", "") or "").strip()
+                return "git checkout %s 失败: %s" % (branch, err or target)
+        return None
+
+    def _prepare_pending_environments(self, limit=5):
+        """准备所有 status=creating 的环境（run_all 每轮开头调）。
+
+        每条环境独立 try：一条失败只把自己标 failed，不挡其它
+        环境和任务。状态写回用 CAS（WHERE status='creating'）：
+        准备期间用户归档了环境时 CAS 改不到行，准备结果就此
+        作废、状态保持 archived，绝不把归档覆盖回 ready。worker
+        中途死了、环境仍是 creating 时下一轮会重新准备（准备
+        本身幂等），没有“preparing 卡死”这种中间态。返回
+        [(env_id, status, error_or_None), ...] 供日志与单测。"""
+        try:
+            conn = _db_connect(self.db_path)
+            try:
+                rows = conn.execute(
+                    "SELECT * FROM environments WHERE status = 'creating'"
+                    " ORDER BY created_at, id LIMIT ?",
+                    (limit,)).fetchall()
+            finally:
+                conn.close()
+        except Exception as exc:
+            log_event(self.logger, "worker_env_list_failed",
+                      message="worker env list failed: %s" % exc)
+            return []
+        results = []
+        for row in rows:
+            try:
+                env = {
+                    "id": row["id"],
+                    "project_id": row["project_id"],
+                    "name": row["name"],
+                    "branch": row["branch"] if "branch" in row.keys()
+                    else None,
+                    "executor": row["executor"] if "executor" in row.keys()
+                    else None,
+                    "repo_url": row["repo_url"] if "repo_url" in row.keys()
+                    else None,
+                }
+            except Exception:
+                continue
+            try:
+                error = self._prepare_environment(env)
+            except Exception as exc:
+                error = str(exc)
+            status = "failed" if error else "ready"
+            try:
+                conn = _db_connect(self.db_path)
+                try:
+                    cur = conn.execute(
+                        "UPDATE environments SET status = ?,"
+                        " error_text = ?, updated_at = ?"
+                        " WHERE id = ? AND status = 'creating'",
+                        (status, error, time.time(), env["id"]))
+                    conn.commit()
+                    if cur.rowcount == 0:
+                        # 准备期间被归档：状态以归档为准，不覆盖。
+                        status = "archived"
+                finally:
+                    conn.close()
+            except Exception as exc:
+                log_event(self.logger, "worker_env_update_failed",
+                          message="worker env update failed: %s" % exc)
+                continue
+            results.append((env["id"], status, error))
+        return results
+
+    def _apply_environment_repo(self, task):
+        """任务带 environment_id 时的派发：环境必须已 ready 且
+        属于本任务项目，执行器冲突时拒绝；通过后 task["repo"]
+        指到环境检出（租户 Docker 为容器内路径），后续执行器
+        与无环境任务完全同款——host.py 从检出 HEAD 为任务建
+        worktree，任务隔离、合并、提 PR 都不变。环境未 ready
+        时返回错误文本，调用方把任务标 failed（仓库解析失败
+        绝不重试，见 _apply_project_repo 的约定）。"""
+        env_id = task.get("environment_id")
+        env = self._fetch_environment(env_id)
+        if env is None:
+            return "environment not found: %s" % env_id
+        if env["project_id"] != task.get("project_id"):
+            return "environment belongs to another project"
+        if env["status"] != "ready":
+            return "environment not ready: %s" % env["status"]
+        task_executor = task.get("executor")
+        if env["executor"] and task_executor \
+                and env["executor"] != task_executor:
+            return "executor does not match environment executor"
+        task["repo"] = self._env_repo_path(env)
+        return None
+
     # -- 项目仓库 ------------------------------------------------------
     def _project_repo_url(self, project_id):
         """查任务所属项目的 repo_url；查不到项目、列缺失、表缺失时
@@ -1132,7 +1354,13 @@ class Worker:
         resolve 失败返回错误文本，调用方据此把任务置 failed
         （仓库解析失败绝不标 completed，也不在此重试，避免对
         不可用的仓库反复 clone 烧时间）。
+
+        任务带 environment_id 时不走项目仓库路径，直接走
+        _apply_environment_repo（环境检出已由 worker 准备好，
+        这里只校验 ready/归属/执行器并指过去，不再 clone）。
         """
+        if task.get("environment_id"):
+            return self._apply_environment_repo(task)
         use_tenant = self.tenant_isolation and \
             isinstance(self.executor, DockerExecutor)
         repo_url = self._project_repo_url(task.get("project_id"))
@@ -1945,6 +2173,15 @@ class Worker:
         停机信号（_loop_stop/stop_event）只阻止新认领，不会
         丢弃已提交的任务；排空等待由 run_loop 的 finally 负责。
         """
+        # 环境准备先跑：每轮先把 creating 的环境准备好（clone +
+        # checkout 分支），再认领普通任务；一条环境失败只标它
+        # 自己 failed，不挡任务——准备函数内部已隔离异常，这里
+        # 再包一层，防 DB 结构异常（如老库缺表）带走整轮任务。
+        try:
+            self._prepare_pending_environments()
+        except Exception as exc:
+            log_event(self.logger, "worker_env_prepare_failed",
+                      message="worker env prepare failed: %s" % exc)
         # pr 任务先跑（HA 阶段 1）：每轮先把本主机的 queued pr 任务
         # 跑完再认领普通任务；pr_jobs_only 时只跑 pr 任务。
         self._run_pr_jobs()