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

环境是项目仓库的命名基准检出（项目 + 仓库快照 + 分支 + 执行器），不是官方 Codex 云端环境（无公开调用口，不宣称能建它）。 - DB：environments 表（creating/ready/failed/archived，部分唯一索引保证同项目未归档名唯一、归档名可重用）+ tasks.environment_id， server/worker/db 三处迁移同步（含 #67 的 goal_mode，不丢新语义）。 - API：POST/GET /environments、GET /environments/{id}、 POST /environments/{id}/archive；任务创建选环境须本项目已 ready、执行器一致或继承、cloud 执行器拒绝。 - Worker：run_all 每轮开头准备 creating 环境（真实 git clone + checkout 分支，幂等、无 preparing 卡死态，准备期归档 CAS 不覆盖）；环境任务派发时 task[repo] 指到环境检出（租户 Docker 容器内路径 + envs 整目录挂载），任务仍一任务一 worktree。 - Console：环境卡（列表/新建/归档，未 ready 置灰）+ 新建任务环境下拉 + 任务列表 [环境] 标记；公网 nginx 须同批代理 /environments。 - 单测：新增 test_task_environments.py 14 个；本地实弹（真实 git clone octocat/Hello-World，不跑模型）项目→环境→任务→归档全通。
2026-10-27 04:07:54 +08:00
1 parent c86b450851
commit 3d45edc8b4
@@ -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):