codex-app-server/cloud/handler_project.py

"""Handler mixin：ProjectMixin。

由 CloudControlPlaneHandler 拆出，方法体逐字搬移；
self.* 调用（含其它 mixin 的方法与留守的 _send_json / _db 等）
在混入后的同一类上解析，行为不变。本模块不 import cloud.server。
"""

from cloud.auth import ROLES as _MEMBER_ROLES
from cloud.db import PostgresIntegrityError as _PgIntegrityError
from cloud.push import (
    delete_subscription as _push_delete,
    list_subscriptions as _push_list,
)
from cloud.peer_messages import row_get as _row_get
from cloud.github_app import (
    list_installations,
)
import re
import sqlite3
import time
import uuid
from cloud.repos import validate_repo_url
from cloud.handler_support import (
    MAX_PROJECTS_PER_USER as MAX_PROJECTS_PER_USER,
    MAX_PROJECT_ID_LEN as MAX_PROJECT_ID_LEN,
    MAX_PROJECT_NAME_LEN as MAX_PROJECT_NAME_LEN,
    _auth_store as _auth_store,
    _environment_to_dict as _environment_to_dict,
)


class ProjectMixin:
    """ProjectMixin 领域方法（见模块 docstring）。"""

    # -- 环境 ----------------------------------------------------------
    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()
        if err:
            self._send_json(400, {"error": err})
            return
        name = body.get("name")
        if not name or not isinstance(name, str):
            self._send_json(400, {"error": "name required"})
            return
        if len(name) > MAX_PROJECT_NAME_LEN:
            self._send_json(400, {
                "error": "project name too long: max %d characters"
                         % MAX_PROJECT_NAME_LEN})
            return
        project_id = body.get("id") or uuid.uuid4().hex
        if not isinstance(project_id, str):
            self._send_json(400, {"error": "id must be a string"})
            return
        if len(project_id) > MAX_PROJECT_ID_LEN:
            self._send_json(400, {
                "error": "project id too long: max %d characters"
                         % MAX_PROJECT_ID_LEN})
            return
        # 单用户项目数上限：按 projects.owner_id 计（成员身份加入
        # 的项目不计入自己的上限，避免“被拉进很多项目就建不了
        # 项目”）；超限 400 且不建项目。
        _count_conn = self._db()
        try:
            _owned = _count_conn.execute(
                "SELECT COUNT(*) FROM projects WHERE owner_id = ?",
                (user_id,)).fetchone()[0]
        finally:
            _count_conn.close()
        if _owned >= MAX_PROJECTS_PER_USER:
            self._send_json(400, {
                "error": "project limit exceeded: max %d projects"
                         " per user" % MAX_PROJECTS_PER_USER})
            return
        # repo_url 可选：不传时为 None（老行为不变）；传了则与 worker
        # 共用 cloud.repos.validate_repo_url 的同一套校验，不合法时
        # 400 且不建项目，绝不把带凭证的 URL 存进库。
        repo_url = body.get("repo_url")
        if repo_url is not None:
            try:
                repo_url = validate_repo_url(repo_url)
            except ValueError as exc:
                self._send_json(400, {"error": str(exc)})
                return
        # auto_pr 可选，默认 0（关闭）：只接受 0/1（布尔 True/False
        # 按 1/0 计），其它值 400 且不建项目。本轮没有现成的
        # PATCH /projects/{id} 路由，auto_pr 只能在创建时指定，
        # 建后改它需直接改库或重建项目（见 README）。
        auto_pr_raw = body.get("auto_pr", 0)
        if isinstance(auto_pr_raw, bool):
            auto_pr = 1 if auto_pr_raw else 0
        elif isinstance(auto_pr_raw, int) and auto_pr_raw in (0, 1):
            auto_pr = auto_pr_raw
        else:
            self._send_json(400, {"error": "auto_pr must be 0 or 1"})
            return
        now = time.time()
        conn = self._db()
        try:
            existing = conn.execute(
                "SELECT id FROM projects WHERE id = ?", (project_id,)
            ).fetchone()
            if existing:
                self._send_json(409, {"error": "project exists"})
                return
            conn.execute(
                "INSERT INTO projects(id, name, owner_id, created_at,"
                " repo_url, auto_pr) VALUES(?, ?, ?, ?, ?, ?)",
                (project_id, name, user_id, now, repo_url, auto_pr),
            )
            conn.commit()
            project = {
                "id": project_id,
                "name": name,
                "owner_id": user_id,
                "created_at": now,
                "repo_url": repo_url,
                "auto_pr": auto_pr,
            }
        except (sqlite3.IntegrityError, _PgIntegrityError):
            self._send_json(409, {"error": "project exists"})
            return
        finally:
            conn.close()
        # AuthStore.create_project 只写成员表（owner 角色），项目记录
        # 本身在上面的 projects 表；两者配合才算一个完整项目。
        with _auth_store(self.server.db_path) as store:
            store.create_project(project_id, user_id)
            store.audit(user_id, project_id, "create_project", name)
        self._send_json(201, project)

    def _update_project(self, user_id, project_id):
        """PATCH /projects/{id}：项目 owner 可改 repo_url（建后补仓
        库，解“建项目时没填仓库、之后建环境一直 400”的死结）。
        repo_url 缺 key 时 400；null 或空串表示清空；字符串走与
        创建时同一套 validate_repo_url 校验（带凭证的 URL 拒绝）。
        只改 projects 行，不动已建环境的快照 repo_url——已建环境
        仍用建环境时的仓库，新环境才用新值。改完记 update_project
        审计。"""
        err = self._check_project(user_id, project_id, "manage_project")
        if err:
            self._send_json(err[0], {"error": err[1]})
            return
        body, jerr = self._read_json()
        if jerr:
            self._send_json(400, {"error": jerr})
            return
        if "repo_url" not in body:
            self._send_json(400, {"error": "repo_url required"})
            return
        repo_url = body.get("repo_url")
        if repo_url is not None and repo_url != "":
            if not isinstance(repo_url, str):
                self._send_json(400, {"error": "repo_url must be a string"})
                return
            try:
                repo_url = validate_repo_url(repo_url)
            except ValueError as exc:
                self._send_json(400, {"error": str(exc)})
                return
        else:
            repo_url = None
        conn = self._db()
        try:
            row = conn.execute(
                "SELECT id, name, owner_id, created_at, repo_url, auto_pr"
                " FROM projects WHERE id = ?",
                (project_id,)).fetchone()
            if row is None:
                self._send_json(404, {"error": "project not found"})
                return
            conn.execute(
                "UPDATE projects SET repo_url = ? WHERE id = ?",
                (repo_url, project_id))
            conn.commit()
            row = conn.execute(
                "SELECT id, name, owner_id, created_at, repo_url, auto_pr"
                " FROM projects WHERE id = ?",
                (project_id,)).fetchone()
        finally:
            conn.close()
        with _auth_store(self.server.db_path) as store:
            store.audit(user_id, project_id, "update_project",
                        repo_url or "(cleared)")
        self._send_json(200, {"project": _project_to_dict(row)})

    def _github_repos_list(self, user_id):
        """GET /github/repos：只返回调用者成员项目对应的仓库。

        P1 收紧前返回全部安装记录（任一已认证用户可看到全部
        安装仓库名，属跨租户信息泄露）。现按 project_members
        过滤：只保留调用者是成员的项目的 repo_url 对应的仓库
        （clone_url 归一化后比对：去 .git 后缀、比对小写）；
        调用者无成员项目或项目未填 repo_url 时返回空列表。
        手填 URL 的能力不变——没进 App 安装的公开仓库仍可手动填。"""
        conn = self._db()
        try:
            rows = list_installations(conn)
            member_ids = {
                row["project_id"]
                for row in conn.execute(
                    "SELECT project_id FROM project_members"
                    " WHERE user_id = ?",
                    (user_id,)).fetchall()
            }
            allowed = set()
            for row in conn.execute(
                    "SELECT id, repo_url FROM projects").fetchall():
                if row["id"] not in member_ids or not row["repo_url"]:
                    continue
                url = str(row["repo_url"]).strip().lower()
                if url.endswith(".git"):
                    url = url[:-4]
                allowed.add(url)
        finally:
            conn.close()
        repos = []
        for r in rows:
            clone_url = "https://github.com/%s.git" % r["repo_full_name"]
            normalized = clone_url.lower()
            if normalized.endswith(".git"):
                normalized = normalized[:-4]
            if normalized not in allowed:
                continue
            repos.append({
                "installation_id": r["installation_id"],
                "account_login": r["account_login"],
                "repo_full_name": r["repo_full_name"],
                "clone_url": clone_url,
            })
        self._send_json(200, {"repos": repos})

    def _list_projects(self, user_id):
        conn = self._db()
        try:
            member_rows = conn.execute(
                "SELECT project_id, role FROM project_members"
                " WHERE user_id = ?",
                (user_id,)).fetchall()
            member_ids = {row["project_id"] for row in member_rows}
            roles = {row["project_id"]: row["role"]
                     for row in member_rows}
            rows = conn.execute(
                "SELECT id, name, owner_id, created_at, repo_url, auto_pr"
                " FROM projects ORDER BY created_at, id"
            ).fetchall()
        finally:
            conn.close()
        projects = []
        for r in rows:
            if r["id"] not in member_ids:
                continue
            # 附加字段 my_role（只增字段，不改已有成功语义）：
            # 控制台据此按角色禁用重试/合并/提 PR 等按钮；服务端
            # 权限仍以每次请求的 check_permission 为准，前端禁用
            # 只是防误点，不是鉴权。
            item = _project_to_dict(r)
            item["my_role"] = roles.get(r["id"])
            projects.append(item)
        self._send_json(200, {"projects": projects})

    # -- 项目成员：邀请 / 列表 / 改角色 -------------------------------
    def _can_manage_members(self, user_id, project_id):
        """成员管理权：项目 owner 或全局管理员。

        全局管理员（CODEX_GLOBAL_ADMINS）不一定是项目成员，
        故不能只用项目内 manage_project 判定；项目 owner 判定
        复用 check_permission 的 manage_project（仅 owner）。
        """
        if self._is_global_admin(user_id):
            return True
        with _auth_store(self.server.db_path) as store:
            return store.check_permission(
                user_id, project_id, "manage_project")

    def _set_member_row(self, store, project_id, actor_id, target_id,
                        role):
        """写成员行并记审计：owner 走 AuthStore.set_member（内含
        owner 校验与审计）；非成员的全局管理员走直接写 + 同格式
        审计（set_member 的 owner 校验会挡住全局管理员，故分流）。
        """
        if store.check_permission(actor_id, project_id,
                                  "manage_project"):
            store.set_member(project_id, actor_id, target_id, role)
            return
        store.conn.execute(
            "INSERT OR REPLACE INTO project_members"
            "(project_id, user_id, role) VALUES(?, ?, ?)",
            (project_id, target_id, role))
        store.conn.commit()
        store.audit(actor_id, project_id, "set_member",
                    "%s:%s" % (target_id, role))

    def _project_members_list(self, user_id, project_id):
        """GET /projects/{id}/members：项目成员（read）或全局
        管理员可读。只返回 user_id / username / role，绝不返回
        key 哈希等认证字段。"""
        if not self._project_exists(project_id):
            self._send_json(404, {"error": "project not found"})
            return
        if not self._is_global_admin(user_id):
            err = self._check_project(user_id, project_id, "read")
            if err:
                self._send_json(err[0], {"error": err[1]})
                return
        conn = self._db()
        try:
            rows = conn.execute(
                "SELECT m.user_id, m.role, u.username FROM"
                " project_members m LEFT JOIN users u"
                " ON u.id = m.user_id WHERE m.project_id = ?"
                " ORDER BY u.username, m.user_id",
                (project_id,)).fetchall()
        finally:
            conn.close()
        self._send_json(200, {"project_id": project_id, "members": [
            {"user_id": r["user_id"], "username": r["username"],
             "role": r["role"]} for r in rows]})

    def _resolve_invite_target(self, body):
        """邀请目标解析：优先 user_id，其次 username 查 id；
        返回 (target_id, error_status, error)。"""
        target = body.get("user_id")
        username = body.get("username")
        if target is not None and not isinstance(target, str):
            return None, 400, "user_id must be a string"
        if username is not None and not isinstance(username, str):
            return None, 400, "username must be a string"
        if not target and not username:
            return None, 400, "user_id or username required"
        conn = self._db()
        try:
            if target:
                row = conn.execute(
                    "SELECT id FROM users WHERE id = ?",
                    (target,)).fetchone()
                if row is None:
                    return None, 404, "user not found"
                return row["id"], None, None
            row = conn.execute(
                "SELECT id FROM users WHERE username = ?",
                (username,)).fetchone()
            if row is None:
                return None, 404, "user not found"
            return row["id"], None, None
        finally:
            conn.close()

    def _project_member_invite(self, user_id, project_id):
        """POST /projects/{id}/members：限项目 owner 或全局管理
        员。已是成员时 409（改角色走 PATCH，不用邀请覆盖）；
        成功记 set_member 审计。role 缺省 contributor。"""
        if not self._project_exists(project_id):
            self._send_json(404, {"error": "project not found"})
            return
        if not self._can_manage_members(user_id, project_id):
            self._send_json(403, {"error": "forbidden"})
            return
        body, jerr = self._read_json()
        if jerr:
            self._send_json(400, {"error": jerr})
            return
        role = body.get("role") or "contributor"
        if role not in _MEMBER_ROLES:
            self._send_json(400, {"error": "unknown role"})
            return
        target, estatus, eerr = self._resolve_invite_target(body)
        if eerr:
            self._send_json(estatus, {"error": eerr})
            return
        with _auth_store(self.server.db_path) as store:
            existing = store.conn.execute(
                "SELECT role FROM project_members"
                " WHERE project_id = ? AND user_id = ?",
                (project_id, target)).fetchone()
            if existing is not None:
                self._send_json(409, {"error": "already_member"})
                return
            self._set_member_row(store, project_id, user_id,
                                 target, role)
        self._send_json(201, {"project_id": project_id,
                              "user_id": target, "role": role})

    def _project_member_patch(self, user_id, project_id, target):
        """PATCH /projects/{id}/members/{user_id}：限项目 owner
        或全局管理员。唯一 owner 降为非 owner 时 409（无论降
        自己还是降他人——否则项目将无 owner 可管）；目标非成员
        时 404。"""
        if not self._project_exists(project_id):
            self._send_json(404, {"error": "project not found"})
            return
        if not self._can_manage_members(user_id, project_id):
            self._send_json(403, {"error": "forbidden"})
            return
        body, jerr = self._read_json()
        if jerr:
            self._send_json(400, {"error": jerr})
            return
        role = body.get("role")
        if role not in _MEMBER_ROLES:
            self._send_json(400, {"error": "unknown role"})
            return
        with _auth_store(self.server.db_path) as store:
            cur = store.conn.execute(
                "SELECT role FROM project_members"
                " WHERE project_id = ? AND user_id = ?",
                (project_id, target)).fetchone()
            if cur is None:
                self._send_json(404, {"error": "member not found"})
                return
            if cur[0] == "owner" and role != "owner":
                owners = store.conn.execute(
                    "SELECT COUNT(*) FROM project_members"
                    " WHERE project_id = ? AND role = 'owner'",
                    (project_id,)).fetchone()[0]
                if owners <= 1:
                    self._send_json(
                        409,
                        {"error": "cannot demote the only owner"})
                    return
            self._set_member_row(store, project_id, user_id,
                                 target, role)
        self._send_json(200, {"project_id": project_id,
                              "user_id": target, "role": role})

    def _project_member_delete(self, user_id, project_id, target):
        """DELETE /projects/{id}/members/{user_id}：限项目 owner
        或全局管理员。删唯一 owner 时 409（否则项目将无 owner
        可管）；目标非成员时 404。成功记 remove_member 审计。"""
        if not self._project_exists(project_id):
            self._send_json(404, {"error": "project not found"})
            return
        if not self._can_manage_members(user_id, project_id):
            self._send_json(403, {"error": "forbidden"})
            return
        with _auth_store(self.server.db_path) as store:
            cur = store.conn.execute(
                "SELECT role FROM project_members"
                " WHERE project_id = ? AND user_id = ?",
                (project_id, target)).fetchone()
            if cur is None:
                self._send_json(404, {"error": "member not found"})
                return
            if cur[0] == "owner":
                owners = store.conn.execute(
                    "SELECT COUNT(*) FROM project_members"
                    " WHERE project_id = ? AND role = 'owner'",
                    (project_id,)).fetchone()[0]
                if owners <= 1:
                    self._send_json(
                        409,
                        {"error": "cannot remove the only owner"})
                    return
            self._remove_member_row(store, project_id, user_id, target)
        self._send_json(200, {"project_id": project_id,
                              "user_id": target})

    def _remove_member_row(self, store, project_id, actor_id, target_id):
        """删成员行并记审计：owner 走 AuthStore.remove_member
        （内含 owner 校验与审计）；非成员的全局管理员走直接删
        + 同格式审计（remove_member 的 owner 校验会挡住全局
        管理员，故分流，与 _set_member_row 同款）。"""
        if store.check_permission(actor_id, project_id,
                                  "manage_project"):
            store.remove_member(project_id, actor_id, target_id)
            return
        store.conn.execute(
            "DELETE FROM project_members"
            " WHERE project_id = ? AND user_id = ?",
            (project_id, target_id))
        store.conn.commit()
        store.audit(actor_id, project_id, "remove_member", target_id)

    # -- 项目子资源：记忆 / 知识库 / 成本 -------------------------------
    def _project_exists(self, project_id):
        conn = self._db()
        try:
            row = conn.execute(
                "SELECT id FROM projects WHERE id = ?", (project_id,)
            ).fetchone()
            return row is not None
        finally:
            conn.close()

    def _check_project(self, user_id, project_id, action):
        """返回 None 表示通过；否则返回 (status, error) 先行响应。"""
        if not self._project_exists(project_id):
            return 404, "project not found"
        with _auth_store(self.server.db_path) as store:
            if not store.check_permission(user_id, project_id, action):
                return 403, "forbidden"
        return None

    def _project_push_subscription_delete(self, user_id, method, project_id, parts):
        # 从 _dispatch_project 平移的 push 订阅删除分支：
        # 语句原样平移，仅缩进减一格；每条路径都已发响应。
        if method != "DELETE":
            self._send_json(404, {"error": "not found"})
            return
        err = self._check_project(user_id, project_id, "create_task")
        if err:
            self._send_json(err[0], {"error": err[1]})
            return
        subs = _push_list(self.server.db_path, project_id)
        mine = [s for s in subs if s["id"] == parts[3]]
        if not mine:
            self._send_json(404, {"error": "subscription not found"})
            return
        # 只能删自己的订阅，项目 owner 可删任意（owner 角色检查）。
        with _auth_store(self.server.db_path) as auth:
            is_owner = auth.check_permission(
                user_id, project_id, "manage_project")
        if mine[0]["user_id"] != user_id and not is_owner:
            self._send_json(403, {"error": "forbidden"})
            return
        deleted = _push_delete(self.server.db_path,
                               subscription_id=parts[3])
        if deleted:
            with _auth_store(self.server.db_path) as store:
                store.audit(
                    user_id, project_id, "push_subscription_deleted",
                    "subscription_id=%s" % parts[3])
        self._send_json(200, {"deleted": bool(deleted)})
        return


def _project_to_dict(row):
    return {
        "id": row["id"],
        "name": row["name"],
        "owner_id": row["owner_id"],
        "created_at": row["created_at"],
        # 老库行缺 repo_url 列时 _row_get 给 None，不炸（列本身由
        # migrate_projects_table 在 init_db 时补，老行读到也是 None）。
        "repo_url": _row_get(row, "repo_url"),
        # auto_pr 默认 0（关闭）：老行缺列时 None 按 0 计，绝不因
        # 迁移就把老项目默认改成自动推。
        "auto_pr": _row_get(row, "auto_pr") or 0,
    }

_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