codex-app-server/cloud/pr.py

"""云端自动 PR：默认关闭、显式触发、缺条件时跳过不失败。

零第三方依赖，只用 Python 标准库；不改 host.py 的任务逻辑，
只复用它的规则：host 任务名 = cloud 任务 id 经安全字符过滤
（见 cloud/sandbox.py LocalExecutor._host_name），分支名是
host/<host 任务名>，worktree 在 <仓库根>/work/host-worktrees/
<host 任务名>。

触发规则（按顺序，任一步跳过即返回，不推、不改任务状态）：
  1. 任务所属项目的 auto_pr 不为 1 -> pr_skipped, auto_pr_disabled
  2. 本地没有对应 worktree -> pr_skipped, no_worktree
  3. worktree 里没有 git remote origin -> pr_skipped, no_remote
  4. 本机没有 gh 二进制 -> pr_skipped, no_gh
  5. gh 未登录（gh auth status 非零） -> pr_skipped,
     gh_not_authenticated
  只有以上全过才 push 分支 + gh pr create；push 或 pr create
  失败时抛 PrStepError（调用方据此返回 502），任务状态保持
  merged 不变，绝不把任务标 failed。

  worker 不自动推：自动 PR 只由显式调 POST /tasks/{id}/pr
  触发。默认自动推会把测试任务也推到远端，本模块不做。

  本模块不存、不打任何密钥：gh 的登录凭证由 gh 自己管
  （gh auth login），push/pr 的命令与错误文本里不出现任何
  token；错误文本只含 git/gh 的原文摘要，不含远端 URL 里的
  userinfo（若 remote URL 含 userinfo，摘要里把它抹掉）。
"""

import os
import re
import shutil
import subprocess

_SKIP_AUTO_PR = "auto_pr_disabled"
_SKIP_NO_WORKTREE = "no_worktree"
_SKIP_NO_REMOTE = "no_remote"
_SKIP_NO_GH = "no_gh"
_SKIP_NOT_AUTHED = "gh_not_authenticated"


class PrSkipped(Exception):
    """跳过：reason 见上列常量，调用方返回 200 + pr_skipped。"""

    def __init__(self, reason):
        super().__init__(reason)
        self.reason = reason


class PrStepError(Exception):
    """push / pr create 哪一步失败：step 为 push 或 pr_create。"""

    def __init__(self, step, message):
        super().__init__(message)
        self.step = step
        self.message = message


def host_name_for_task(task_id):
    """cloud 任务 id -> host 任务名（与 sandbox._host_name 同规则）。"""
    return re.sub(r"[^A-Za-z0-9_.-]", "-", str(task_id))


def default_worktrees_root():
    """默认 worktree 根：<仓库根>/work/host-worktrees。"""
    here = os.path.dirname(os.path.abspath(__file__))
    return os.path.join(os.path.dirname(here), "work", "host-worktrees")


def resolve_worktrees_root(override=None):
    """worktree 根优先级：显式参数 > 环境变量 > 默认。"""
    if override:
        return override
    return os.environ.get("CLOUD_HOST_WORKTREES_DIR") or \
        default_worktrees_root()


def _scrub_url(text):
    """抹掉 URL 里的 userinfo（https://user:pass@host -> https://host），
    防止错误文本把凭证带出去。"""
    if not text:
        return text
    return re.sub(r"(https?://)[^/@\s]+@", r"\1", text)


def _run(cmd, cwd=None, runner=None):
    """命令唯一入口：runner 供单测注入，返回带 returncode/stdout/
    stderr 的对象；真实路径走 subprocess。"""
    if runner is not None:
        return runner(cmd, cwd)
    return subprocess.run(cmd, cwd=cwd, capture_output=True, text=True)


def _git_remote_url(worktree_dir, git_runner=None):
    r = _run(["git", "-C", worktree_dir, "remote", "get-url", "origin"],
             cwd=worktree_dir, runner=git_runner)
    url = (getattr(r, "stdout", "") or "").strip()
    if getattr(r, "returncode", 1) != 0 or not url:
        return None
    return url


def check_pr_ready(worktree_dir, git_runner=None, gh_which=None,
                   gh_runner=None):
    """按顺序做跳过检查，全过返回 (worktree_dir, branch)；任一步
    不过抛 PrSkipped。branch 由调用方按 host 任务名算好传进来
    的是 worktree 检查——本函数只查 worktree 与 remote/gh/登录，
    分支名由 create_task_pr 组装。"""
    if not os.path.isdir(worktree_dir):
        raise PrSkipped(_SKIP_NO_WORKTREE)
    if _git_remote_url(worktree_dir, git_runner) is None:
        raise PrSkipped(_SKIP_NO_REMOTE)
    gh_bin = gh_which if gh_which is not None else shutil.which("gh")
    if not gh_bin:
        raise PrSkipped(_SKIP_NO_GH)
    r = _run(["gh", "auth", "status"], cwd=worktree_dir, runner=gh_runner)
    if getattr(r, "returncode", 1) != 0:
        raise PrSkipped(_SKIP_NOT_AUTHED)
    return True


def create_task_pr(task_id, auto_pr, worktrees_root=None,
                   git_runner=None, gh_runner=None, gh_which=None):
    """给一个已 merged 的云端任务建 PR。

    auto_pr 不为 1 时直接抛 PrSkipped(auto_pr_disabled)，调用方
    在查 worktree 之前就应先查项目 auto_pr——本函数也再查一次，
    防止直接调用者漏查。成功返回 {pr_url, pr_number, branch}。
    """
    if auto_pr != 1:
        raise PrSkipped(_SKIP_AUTO_PR)
    host_name = host_name_for_task(task_id)
    worktree_dir = os.path.join(
        resolve_worktrees_root(worktrees_root), host_name)
    branch = "host/" + host_name
    check_pr_ready(worktree_dir, git_runner=git_runner,
                   gh_which=gh_which, gh_runner=gh_runner)
    push = _run(["git", "-C", worktree_dir, "push", "origin", branch],
                cwd=worktree_dir, runner=git_runner)
    if getattr(push, "returncode", 1) != 0:
        raise PrStepError(
            "push", "git push origin %s 失败：%s" % (
                branch, _scrub_url(
                    (getattr(push, "stderr", "") or "")
                    or (getattr(push, "stdout", "") or ""))))
    title = "cloud: %s" % task_id
    r = _run(["gh", "pr", "create", "--title", title,
              "--body", "cloud 任务 %s 的自动 PR。" % task_id,
              "--head", branch],
             cwd=worktree_dir, runner=gh_runner)
    out = (getattr(r, "stdout", "") or "").strip()
    if getattr(r, "returncode", 1) != 0 or not out:
        raise PrStepError(
            "pr_create", "gh pr create 失败：%s" % _scrub_url(
                (getattr(r, "stderr", "") or "") or out))
    pr_url = out.splitlines()[0].strip()
    m = re.search(r"/(\d+)\s*$", pr_url)
    pr_number = int(m.group(1)) if m else None
    return {"pr_url": pr_url, "pr_number": pr_number, "branch": branch}