Merge pull request #39 from openqj/app-token

App-token：推分支建 PR 改走 GitHub App

cloud/README.md

@@ -172,7 +172,7 @@ client_secret（只走环境变量），不是无认证后门。
   没有返回 404）后恢复原来的 oauth 账号登录。查看自己的绑定：
   `GET /oauth/{provider}/bindings`（须持本站 key，只看得到自己的）。
 
-### GitHub App 一键安装（已实现安装一半；App-token API 动作未做）
+### GitHub App 一键安装与 App-token（安装 + token 均已实现）
 
 目标：把手动建 webhook（填 URL、secret、选事件）变成“点一下
 安装”。流程用 GitHub manifest 流程（GitHub 文档标准流程）：
@@ -207,9 +207,18 @@ client_secret（只走环境变量），不是无认证后门。
 /fix-review 也有内容去重（repo + PR + 评论 id，
 `github_fix_review_dedup` 表），只建一次任务。
 
-诚实边界：pem 虽然存下来了，但本阶段不用它签 JWT、不用它换
-installation token；推分支、建 PR 仍走 gh + PAT（见“云端派发与
-自动 PR”），App-token API 动作是下一阶段。
+诚实边界（已缩小）：pem 现在用于签 JWT、换 installation
+token：`cloud/github_app.py` 的 `mint_app_jwt` 是纯标准库
+RS256（解析 PKCS#1 / PKCS#8 私钥，不依赖 openssl、不落临时
+文件），`exchange_installation_token` 换 1 小时有效的 token
+（现换现用、不缓存、不存库），`resolve_app_token` 按仓库名
+查 `github_app_installations` 找 installation id，任一步缺失
+或失败返回 None（调用方回落 PAT，不报错）。推分支、建 PR
+走 App 还是 PAT 见“云端派发与自动 PR”；host.py 本地操作
+（pr / checks / merge）仍是操作员自己的 gh + PAT，不走
+App-token。App 安装事件的现场记录路径仍待下一次真实
+安装/增减仓库动作实弹验证（此前一次是数据修复，不是
+webhook 实测）。
 
 ### 推送凭证清单
 
@@ -830,8 +839,17 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳
   才先把 worktree 改动提交到分支上（worktree 必须正好在
   host/<host 任务名> 上，否则直接失败；nothing to commit 时跳过；
   缺 git 身份时回退 cloud/cloud@local 再试），再 push 分支 +
-  gh pr create，成功后把 pr_url/pr_number 写回任务
-  记录并记审计 pr_created，返回 200 含 pr_url。push 或 pr create
+  建 PR，成功后把 pr_url/pr_number 写回任务
+  记录并记审计 pr_created，返回 200 含 pr_url。App-token
+  路径：项目 repo_url 能解析出 github.com 仓库名且库里有该
+  仓库的 App 安装记录时，端点现换 installation token 走 App
+  路径（响应里 auth_via=app）：push 用 git credential helper，
+  token 只进子进程环境变量 GIT_TOKEN（不进命令行参数、不进
+  文件，helper 命令里只有变量名），建 PR 走 GitHub REST API
+  （先读仓库 default_branch 当 base，不猜 main）；没有安装
+  记录或换 token 失败时自动回落原来的 gh + PAT 路径
+  （auth_via=pat），不因此报错；App 没装的仓库与 gh 路径
+  的行为与原来完全一致。push 或 pr create
   失败时返回 502 且错误文本写清是哪一步（commit、push 还是
   pr_create），
   任务状态保持 merged 不变，绝不把任务标 failed。顺序：任务
@@ -846,7 +864,8 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳
   的路由。gh 的登录凭证由 gh 自己管，控制面不存、不打任何 token。
 - 仍未做（不要宣称已做）：厂商专属适配（ECS、Cloud Run 等，需
   用户选厂商后另做）；默认自动推（本轮明确不做）；主机上 gh 的安装
-  与登录、远端写凭证（控制面不存，自动 PR 的真实成功路径仍需它们）；
+  与登录、远端写凭证（控制面不存；PAT 路径的自动 PR 真实成功
+  仍需它们，App-token 路径不需要 gh）；
   多 worker 的并发容量 enforcement（capacity 目前只是注册表展示，
   run_loop 仍串行）；worker 经 HTTP 发心跳的凭证供给（worker 直接
   写库时不需要，外部 worker 走 HTTP 时需自带 API key）。

cloud/github_app.py

@@ -30,6 +30,7 @@ env secret，App webhook 用库里 secret，验签时两个都试（过渡
 注入 fake http 客户端做单测。
 """
 
+import hashlib
 import json
 import secrets
 import time
@@ -410,3 +411,224 @@ def set_fix_review_task_id(conn, repo, pr_number, comment_id, task_id):
         (task_id, repo, pr_number, str(comment_id)),
     )
     conn.commit()
+
+
+# -- App-token 阶段：JWT 与 installation token ---------------------------
+# 纯标准库 RS256：解析 PEM（PKCS#1 / PKCS#8）取 (n, e, d)，
+# PKCS#1 v1.5 + SHA-256 签名。不依赖 openssl CLI、不落临时文件；
+# JWT 与 installation token 只在进程内出现，绝不打印、不进日志。
+# GitHub App 的 pem 格式：manifest 换回的通常是 PKCS#1
+#（"BEGIN RSA PRIVATE KEY"），openssl 3 默认生成的是 PKCS#8
+#（"BEGIN PRIVATE KEY"），两者都支持。
+
+import base64 as _base64
+
+_SHA256_DIGESTINFO_PREFIX = bytes.fromhex(
+    "3031300d060960864801650304020105000420")
+
+
+def _b64url(data):
+    if isinstance(data, str):
+        data = data.encode("utf-8")
+    return _base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
+
+
+def _der_read(data, pos):
+    """读一个 DER TLV：返回 (tag, value, next_pos)。"""
+    tag = data[pos]
+    pos += 1
+    length = data[pos]
+    pos += 1
+    if length & 0x80:
+        nbytes = length & 0x7F
+        if nbytes == 0 or nbytes > 4:
+            raise AppError("unsupported DER length",
+                           reason="bad_key")
+        length = int.from_bytes(data[pos:pos + nbytes], "big")
+        pos += nbytes
+    return tag, data[pos:pos + length], pos + length
+
+
+def _der_sequence_items(der):
+    """SEQUENCE 的子项列表（每项为 (tag, value)）。"""
+    tag, value, _ = _der_read(der, 0)
+    if tag != 0x30:
+        raise AppError("expected DER SEQUENCE", reason="bad_key")
+    items = []
+    pos = 0
+    while pos < len(value):
+        t, v, pos = _der_read(value, pos)
+        items.append((t, v))
+    return items
+
+
+def _parse_pkcs1_der(der):
+    """PKCS#1 RSAPrivateKey DER -> (n, e, d)。"""
+    items = _der_sequence_items(der)
+    if len(items) < 4:
+        raise AppError("short PKCS#1 key", reason="bad_key")
+    n = int.from_bytes(items[1][1], "big")
+    e = int.from_bytes(items[2][1], "big")
+    d = int.from_bytes(items[3][1], "big")
+    if n <= 0 or e <= 0 or d <= 0:
+        raise AppError("invalid RSA parameters", reason="bad_key")
+    return n, e, d
+
+
+def _parse_rsa_private_key(pem):
+    """解析 PEM 私钥 -> (n, e, d)，PKCS#1 与 PKCS#8 都支持。"""
+    if not pem or not isinstance(pem, str):
+        raise AppError("private key required", reason="bad_key")
+    body = "".join(
+        line.strip() for line in pem.splitlines()
+        if line.strip() and not line.strip().startswith("-----"))
+    try:
+        der = _base64.b64decode(body)
+    except Exception as exc:
+        raise AppError("private key is not valid base64: %s" % (exc,),
+                       reason="bad_key") from exc
+    if "RSA PRIVATE KEY" in pem:
+        return _parse_pkcs1_der(der)
+    if "PRIVATE KEY" in pem:
+        items = _der_sequence_items(der)
+        if len(items) < 3 or items[2][0] != 0x04:
+            raise AppError("unexpected PKCS#8 shape", reason="bad_key")
+        return _parse_pkcs1_der(items[2][1])
+    raise AppError("unsupported private key format", reason="bad_key")
+
+
+def rsa_sign_sha256(pem, message):
+    """RS256 签名：返回签名 bytes（长度 = 模长字节数）。"""
+    n, _e, d = _parse_rsa_private_key(pem)
+    if isinstance(message, str):
+        message = message.encode("utf-8")
+    digest = hashlib.sha256(message).digest()
+    t = _SHA256_DIGESTINFO_PREFIX + digest
+    em_len = (n.bit_length() + 7) // 8
+    if em_len < len(t) + 11:
+        raise AppError("RSA key too small for SHA-256", reason="bad_key")
+    em = b"\x00\x01" + b"\xff" * (em_len - len(t) - 3) + b"\x00" + t
+    m = int.from_bytes(em, "big")
+    s = pow(m, d, n)
+    return s.to_bytes(em_len, "big")
+
+
+def mint_app_jwt(app_id, pem, now=None, ttl_seconds=540):
+    """签发 App JWT：header alg=RS256，有效期默认 9 分钟
+    （GitHub 上限 10 分钟，iat 回拨 60 秒防时钟 skew）。
+
+    now 只为单测注入时间。pem 非法时抛 AppError（调用方据此
+    回落 PAT，不静默用坏 token 去撞 GitHub）。
+    """
+    if app_id is None or str(app_id) == "":
+        raise AppError("app_id required", reason="bad_key")
+    if now is None:
+        now = time.time()
+    header = _b64url(json.dumps({"alg": "RS256", "typ": "JWT"},
+                                 separators=(",", ":")))
+    payload = _b64url(json.dumps(
+        {"iat": int(now) - 60, "exp": int(now) + int(ttl_seconds),
+         "iss": str(app_id)},
+        separators=(",", ":")))
+    signing_input = header + "." + payload
+    signature = rsa_sign_sha256(pem, signing_input.encode("ascii"))
+    return signing_input + "." + _b64url(signature)
+
+
+def _github_api_request(method, url, token, payload, http_client):
+    """调 GitHub API：返回 (status, dict)。
+
+    http_client 注入约定（单测用）：``http_client(method, url,
+    headers, body_bytes_or_None) -> (status, dict)``。真实路径
+    用 urllib，HTTPError 也解析 body（GitHub 的错误体是 JSON，
+    调用方按 status 判断，不靠异常）。
+    """
+    headers = {"Accept": "application/vnd.github+json",
+               "User-Agent": "codex-app-server"}
+    if token:
+        headers["Authorization"] = "Bearer " + token
+    body = None
+    if payload is not None:
+        body = json.dumps(payload).encode("utf-8")
+    if http_client is not None:
+        result = http_client(method, url, headers, body)
+        if not (isinstance(result, tuple) and len(result) == 2):
+            raise AppError("http_client must return (status, dict)",
+                           reason="token_exchange_failed")
+        return result
+    request = urllib.request.Request(url, data=body, method=method,
+                                     headers=headers)
+    try:
+        with urllib.request.urlopen(request, timeout=30) as response:
+            raw = response.read().decode("utf-8", "replace")
+            status = response.status
+    except urllib.error.HTTPError as e:
+        raw = e.read().decode("utf-8", "replace")
+        status = e.code
+    try:
+        parsed = json.loads(raw)
+    except ValueError:
+        parsed = {}
+    if not isinstance(parsed, dict):
+        parsed = {}
+    return status, parsed
+
+
+def exchange_installation_token(app_id, pem, installation_id,
+                                http_client=None, now=None):
+    """换 installation token：返回 {token, expires_at, permissions,
+    repository_selection}。token 有效期 1 小时，调用方现换现用，
+    不缓存、不存库。换失败抛 AppError（调用方回落 PAT）。"""
+    if not installation_id:
+        raise AppError("installation_id required",
+                       reason="token_exchange_failed")
+    jwt = mint_app_jwt(app_id, pem, now=now)
+    url = ("https://api.github.com/app/installations/%s/access_tokens"
+           % installation_id)
+    status, data = _github_api_request("POST", url, jwt, None, http_client)
+    if status != 201 or not data.get("token"):
+        raise AppError(
+            "installation token exchange failed: %s %r"
+            % (status, data.get("message")),
+            reason="token_exchange_failed")
+    return {"token": data["token"], "expires_at": data.get("expires_at"),
+            "permissions": data.get("permissions"),
+            "repository_selection": data.get("repository_selection")}
+
+
+def find_installation_for_repo(conn, repo_full_name):
+    """查仓库对应的 App installation id：无记录返回 None
+    （调用方回落 PAT，不报错——没装 App 的仓库走老路）。"""
+    if not repo_full_name:
+        return None
+    _ensure_installations_table(conn)
+    row = conn.execute(
+        "SELECT installation_id FROM github_app_installations"
+        " WHERE repo_full_name = ? ORDER BY created_at LIMIT 1",
+        (repo_full_name,),
+    ).fetchone()
+    return row[0] if row else None
+
+
+def resolve_app_token(conn, repo_full_name, http_client=None):
+    """给仓库解析 App installation token：成功返回 token 字符串，
+    任一步缺失或失败返回 None（调用方回落 PAT，不抛错）。
+
+    步骤：App 配置（库里）-> 安装记录（仓库名查 installation
+    id）-> JWT -> 换 token。token 与 JWT 只在进程内出现。
+    """
+    try:
+        config = get_app_config(conn)
+        if config is None or not config.get("pem"):
+            return None
+        installation_id = find_installation_for_repo(conn, repo_full_name)
+        if installation_id is None:
+            return None
+        result = exchange_installation_token(
+            config.get("app_id"), config.get("pem"), installation_id,
+            http_client=http_client)
+        return result["token"]
+    except AppError:
+        return None
+    except Exception:
+        return None

cloud/pr.py

@@ -38,6 +38,15 @@ host/<host 任务名>，worktree 在 <仓库根>/work/host-worktrees/
   （gh auth login），push/pr 的命令与错误文本里不出现任何
   token；错误文本只含 git/gh 的原文摘要，不含远端 URL 里的
   userinfo（若 remote URL 含 userinfo，摘要里把它抹掉）。
+
+  App-token 路径（cloud/github_app.py 的 resolve_app_token 供
+  token）：调用方传 app_token 时，跳过 gh 二进制与 gh 登录
+  检查（App 路径不需要 gh）；push 走 git credential helper，
+  token 只放进子进程环境变量 GIT_TOKEN，不进命令行参数、不
+  进文件，helper 命令里只有变量名；建 PR 走 GitHub REST API
+  （先读仓库 default_branch 再建，base 不猜 main）。没传
+  app_token 时走原来的 gh 路径，两条路不互相干扰；App 没装
+  在该仓库时调用方不传 token、自动回落 PAT。
 """
 
 import os
@@ -96,12 +105,69 @@ def _scrub_url(text):
     return re.sub(r"(https?://)[^/@\s]+@", r"\1", text)
 
 
-def _run(cmd, cwd=None, runner=None):
+def _run(cmd, cwd=None, runner=None, env=None):
     """命令唯一入口：runner 供单测注入，返回带 returncode/stdout/
-    stderr 的对象；真实路径走 subprocess。"""
+    stderr 的对象；真实路径走 subprocess。env 只在真实路径生效
+    （与进程环境合并），注入的 runner 不收 env——token 绝不进
+    假 runner 记录的命令，也不进真实路径的命令行参数。"""
     if runner is not None:
         return runner(cmd, cwd)
-    return subprocess.run(cmd, cwd=cwd, capture_output=True, text=True)
+    full_env = None
+    if env:
+        full_env = dict(os.environ)
+        full_env.update(env)
+    return subprocess.run(cmd, cwd=cwd, capture_output=True, text=True,
+                          env=full_env)
+
+
+_GITHUB_REMOTE_RE = re.compile(
+    r"^(?:https?://(?:[^/@\s]+@)?github\.com/"
+    r"|git@github\.com:|ssh://git@github\.com/)"
+    r"([^/\s]+)/([^/\s]+?)(?:\.git)?/?$")
+
+
+def parse_git_remote_owner_repo(url):
+    """解析 git remote URL -> (owner, repo)，非 github.com 仓库
+    或解析失败返回 None（调用方据此回落 PAT，不报错）。"""
+    if not url or not isinstance(url, str):
+        return None
+    m = _GITHUB_REMOTE_RE.match(url.strip())
+    if not m:
+        return None
+    return m.group(1), m.group(2)
+
+
+def _default_github_api(method, url, token, payload):
+    """默认 GitHub API 客户端：复用 github_app 的 urllib 实现
+    （返回 (status, dict)；懒 import，避免模块导入环）。"""
+    from cloud.github_app import _github_api_request
+    return _github_api_request(method, url, token, payload, None)
+
+
+def _create_pr_via_api(owner, repo, branch, title, body, token,
+                       github_api):
+    """用 installation token 走 REST API 建 PR：先读仓库
+    default_branch 当 base（不猜 main），再 POST /pulls。
+    成功返回 {pr_url, pr_number, branch}，失败抛 PrStepError。"""
+    api = github_api or _default_github_api
+    status, repo_info = api(
+        "GET", "https://api.github.com/repos/%s/%s" % (owner, repo),
+        token, None)
+    if status != 200:
+        raise PrStepError(
+            "pr_create",
+            "GitHub API 读仓库失败：%s" % (repo_info.get("message"),))
+    base = repo_info.get("default_branch") or "main"
+    status, data = api(
+        "POST", "https://api.github.com/repos/%s/%s/pulls" % (owner, repo),
+        token, {"title": title, "head": branch, "base": base,
+                "body": body})
+    if status not in (200, 201) or not data.get("html_url"):
+        raise PrStepError(
+            "pr_create",
+            "GitHub API 建 PR 失败：%s" % (data.get("message"),))
+    return {"pr_url": data["html_url"], "pr_number": data.get("number"),
+            "branch": branch}
 
 
 def _git_remote_url(worktree_dir, git_runner=None):
@@ -114,15 +180,18 @@ def _git_remote_url(worktree_dir, git_runner=None):
 
 
 def check_pr_ready(worktree_dir, git_runner=None, gh_which=None,
-                   gh_runner=None):
+                   gh_runner=None, app_token=None):
     """按顺序做跳过检查，全过返回 (worktree_dir, branch)；任一步
     不过抛 PrSkipped。branch 由调用方按 host 任务名算好传进来
     的是 worktree 检查——本函数只查 worktree 与 remote/gh/登录，
-    分支名由 create_task_pr 组装。"""
+    分支名由 create_task_pr 组装。传了 app_token 时跳过 gh
+    二进制与 gh 登录检查（App 路径不需要 gh）。"""
     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)
+    if app_token:
+        return True
     gh_bin = gh_which if gh_which is not None else shutil.which("gh")
     if not gh_bin:
         raise PrSkipped(_SKIP_NO_GH)
@@ -195,7 +264,8 @@ def _commit_worktree_changes(worktree_dir, branch, task_id,
 
 
 def create_task_pr(task_id, auto_pr, worktrees_root=None,
-                   git_runner=None, gh_runner=None, gh_which=None):
+                   git_runner=None, gh_runner=None, gh_which=None,
+                   app_token=None, github_api=None):
     """给一个已 merged 的云端任务建 PR。
 
     auto_pr 不为 1 时直接抛 PrSkipped(auto_pr_disabled)，调用方
@@ -203,6 +273,9 @@ def create_task_pr(task_id, auto_pr, worktrees_root=None,
     防止直接调用者漏查。成功返回 {pr_url, pr_number, branch}。
     全部跳过检查通过后，先把 worktree 改动提交到分支再 push，
     否则未提交的改动进不了 PR。
+    传了 app_token 时走 App 路径：push 用 credential helper
+    （token 只进子进程环境，不进参数），建 PR 走 GitHub API；
+    不传时走原来的 gh 路径。
     """
     if auto_pr != 1:
         raise PrSkipped(_SKIP_AUTO_PR)
@@ -211,9 +284,35 @@ def create_task_pr(task_id, auto_pr, worktrees_root=None,
         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)
+                   gh_which=gh_which, gh_runner=gh_runner,
+                   app_token=app_token)
     _commit_worktree_changes(worktree_dir, branch, task_id,
                              git_runner=git_runner)
+    title = "cloud: %s" % task_id
+    body = "cloud 任务 %s 的自动 PR。" % task_id
+    if app_token:
+        remote_url = _git_remote_url(worktree_dir, git_runner)
+        owner_repo = parse_git_remote_owner_repo(remote_url)
+        if owner_repo is None:
+            raise PrStepError(
+                "push",
+                "worktree remote 不是 github.com 仓库，App token 无法"
+                " push：%s" % _scrub_url(remote_url))
+        push = _run(
+            ["git", "-C", worktree_dir, "-c",
+             "credential.helper=!f() { echo username=x-access-token;"
+             " echo password=$GIT_TOKEN; }; f",
+             "push", "origin", branch],
+            cwd=worktree_dir, runner=git_runner,
+            env={"GIT_TOKEN": app_token})
+        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 ""))))
+        return _create_pr_via_api(owner_repo[0], owner_repo[1], branch,
+                                  title, body, app_token, github_api)
     push = _run(["git", "-C", worktree_dir, "push", "origin", branch],
                 cwd=worktree_dir, runner=git_runner)
     if getattr(push, "returncode", 1) != 0:
@@ -222,9 +321,8 @@ def create_task_pr(task_id, auto_pr, worktrees_root=None,
                 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,
+              "--body", body,
               "--head", branch],
              cwd=worktree_dir, runner=gh_runner)
     out = (getattr(r, "stdout", "") or "").strip()

cloud/server.py

@@ -59,7 +59,8 @@ from cloud.db import PostgresIntegrityError as _PgIntegrityError
 from cloud.db import PostgresOperationalError as _PgOperationalError
 from cloud.kb import KBStore
 from cloud.memory import MemoryStore
-from cloud.pr import PrSkipped, PrStepError, create_task_pr
+from cloud.pr import (PrSkipped, PrStepError, create_task_pr,
+                      parse_git_remote_owner_repo)
 from cloud.workers import (
     heartbeat_worker as _registry_heartbeat,
     list_workers as _registry_list,
@@ -100,6 +101,7 @@ from cloud.github_app import (
     record_installation_repos,
     remove_installation,
     remove_installation_repos,
+    resolve_app_token,
     save_app_config,
     set_fix_review_task_id,
 )
@@ -1505,6 +1507,11 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         pr_skipped 与 reason，任务状态保持 merged 不变；push/pr
         创建失败时返回 502 并写清哪一步失败，任务同样保持 merged。
         成功时把 pr_url/pr_number 写回任务并记审计 pr_created。
+        App-token：项目 repo_url 能解析出 github.com 仓库名且
+        库里有该仓库的 App 安装记录时，现换 installation token
+        走 App 路径（响应里 auth_via=app），否则走原来的 gh +
+        PAT 路径（auth_via=pat）；换 token 任一步失败都回落
+        PAT，不因此报错。token 只在本次请求内存里出现。
         """
         row = self._fetch_task(task_id)
         if row is None:
@@ -1525,16 +1532,39 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         conn = self._db()
         try:
             prow = conn.execute(
-                "SELECT auto_pr FROM projects WHERE id = ?",
+                "SELECT auto_pr, repo_url FROM projects WHERE id = ?",
                 (task["project_id"],)).fetchone()
         finally:
             conn.close()
         auto_pr = 0
+        repo_url = None
         if prow is not None:
             try:
                 auto_pr = prow["auto_pr"] or 0
             except (KeyError, IndexError, TypeError):
                 auto_pr = 0
+            try:
+                repo_url = prow["repo_url"]
+            except (KeyError, IndexError, TypeError):
+                repo_url = None
+        app_token = None
+        auth_via = "pat"
+        repo_full_name = None
+        if repo_url:
+            parsed = parse_git_remote_owner_repo(repo_url)
+            if parsed:
+                repo_full_name = "%s/%s" % parsed
+        if repo_full_name:
+            conn = self._db()
+            try:
+                app_token = resolve_app_token(
+                    conn, repo_full_name,
+                    http_client=getattr(
+                        self.server, "pr_app_token_client", None))
+            finally:
+                conn.close()
+            if app_token:
+                auth_via = "app"
         try:
             result = create_task_pr(
                 task_id, auto_pr,
@@ -1542,7 +1572,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                     self.server, "host_worktrees_dir", None),
                 git_runner=getattr(self.server, "pr_git_runner", None),
                 gh_runner=getattr(self.server, "pr_gh_runner", None),
-                gh_which=getattr(self.server, "pr_gh_which", None))
+                gh_which=getattr(self.server, "pr_gh_which", None),
+                app_token=app_token,
+                github_api=getattr(self.server, "pr_github_api", None))
         except PrSkipped as exc:
             self._send_json(200, {
                 "task_id": task_id, "status": task["status"],
@@ -1575,7 +1607,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                         result["pr_url"])
         self._send_json(200, dict(task, pr_url=result["pr_url"],
                                   pr_number=result["pr_number"],
-                                  branch=result["branch"]))
+                                  branch=result["branch"],
+                                  auth_via=auth_via))
 
     def _heartbeat_task(self, user_id, task_id):
         row = self._fetch_task(task_id)

cloud/tests/test_app_token.py

@@ -0,0 +1,244 @@
+"""App-token 单测：纯标准库 RS256 签名、JWT、installation token
+换取、仓库安装查找、resolve 回落。
+
+fixture 是测试专用 RSA key（openssl 现生成、仅供单测，从不用
+于生产；即使公开也无害——它不对应任何真实 GitHub App）。
+不联网：换 token 挂 fake http 客户端。
+"""
+
+import base64
+import hashlib
+import json
+import sys
+import tempfile
+import os
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.db import connect as _db_connect
+    from cloud.github_app import (
+        AppError,
+        _parse_rsa_private_key,
+        exchange_installation_token,
+        find_installation_for_repo,
+        mint_app_jwt,
+        record_installation_repos,
+        resolve_app_token,
+        rsa_sign_sha256,
+        save_app_config,
+    )
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.db import connect as _db_connect
+    from cloud.github_app import (
+        AppError,
+        _parse_rsa_private_key,
+        exchange_installation_token,
+        find_installation_for_repo,
+        mint_app_jwt,
+        record_installation_repos,
+        resolve_app_token,
+        rsa_sign_sha256,
+        save_app_config,
+    )
+
+# 测试专用 RSA 私钥（PKCS#1 与 PKCS#8，同一密钥对的两版编码
+# 是两把不同 key——各测各的格式解析，不要求同模）。
+TEST_KEY_PKCS1 = """
+-----BEGIN RSA PRIVATE KEY-----
+MIIEogIBAAKCAQEA2M/hYSQQAJxQfcLxcRob7cq4miMLbyWv+Diq4qppENU8XvNS
+7p/JSc8XqWRcGX6FusObKjpskakEL1+tf2wBYGW5jbtgiHzg4J46p+iK/OFi9t+8
+O6wjva3+5kWrjvkQIKL/pbGfdvVQjq1j5bg2fbXKbQqb7CbbhxDhRrW3xVt5A+PF
+G3+iP0qHcBvWuN1CIUHVvUORSdkY44+8xCs+RjalMqVsOyPzTuMu+ypfPtaHfcGM
+DWwkvpXUwq9VrszfrGqi3qWJ6nBQy20eXRUeEnorDS7Rolj+/bILhtbUUT/wFzX7
+F/er1zquAQSntgiMUutB50zpPefNe5Jf/uknZwIDAQABAoIBAEK9izVp8nsdozFz
+ZvtX/Czw8G6SRxzJzydr2j64EmuaGGFqKadHDKFp3HDKcgdvCR96wzL2lSEplB45
+J6FH0Ayqg7n7ajXZqye5zXiOfuplPx+bs8fyr3XeL88Wtz7lsG87fdK1oM7vqQKi
+oVO/pxZ/dXang2Poej9TlOU9XvqjIoTo4HNckRhEF7MJV+2nQB/d+edlSt9oLrN4
+UxOEBeTEomj+ujJBWFOasOcztRXMl/cQp3PdSZ5Ap4aXElblK5/MocM+tyh6RU++
+hLVBpxxjH2CQbG+HMn0rxn7NbtpBdYU3GTjBdKJUv0MQ9m8s39AFfX4h1YW7PCte
+uO0tyfECgYEA+pwlgTWnQL5dlwkw+qsg4mhiG0XcDFsfrh6W/TcsUJnfQIGZCJnC
+9NxCJEmfQTxgJWEmTvPj+Mi4iFmoMYlfTo1OqI36pXNnH0fmh4l0l9COAUsarQVJ
+DzDh5LaCpOn8uSfgJSzdoBY4aYkxJfl8AVsApCmHBojp9Xjw1gis1csCgYEA3Xmk
+qm1BeGkUU/nGDCjz6AuIYEbWuj7OwnV/EsvsGsMTDryoWsk0ZTI0KJdyI/RhjMdM
+jnv37EOy8YL1NjD01NinQ7Ac1wY1DSUrA+jJ6ctSkjkPFHKVBmiC2c5TJiTXWxd6
+Ms7fGQesIrdSdbr7+GrkogKCJyKEwskcRNsFIVUCgYAVqKkVIa3yyoEMLJ8vCtss
+thjEgqUh7D4wrjQebkZGMQY6bPjs5V0UDHUq7bH10WH92QeWiz2IotBP+G+w2E6E
+W3IQ3zRqWC7xZSVnUVniRQQA2JzVlz9ZMMBVI/Dmj9JnriqRC++bhlAZzuGQhp9Q
+/qxvg6dHyU1vCaTG2rkVQQKBgGrkviwfhVznYo9RW1wdG/OgNVc6eYL0s3g5m3Zo
+A/G6u2T+e0cuG2LIQ1JgHg/qcHaw7mtKvBNFbYWRTma0ZEaitsA9rP2/GjXY/ZTq
+cyGrJ5Q5g+S7lN+964HSSmF7oJFvmGQTYqxC/rrWvi7ia6CvfuAO0+CtkWQuCr3U
+8/ntAoGAawnmD9G1JVUSo1ONbV2gsg/Lv6SSq0dbWja4B0fhAR4BWtSFQV8ePTwk
+Q5bSZc/zdNHcKPIMr29Aa3TTy7jXHkAs2frAwKGRyErt/cVAbah+BLOWxniTNfgD
+6CvUI8gTLJ+Dklmer2o6b0JZEeSxz93xx7sy9E5SqHGStOZRutA=
+-----END RSA PRIVATE KEY-----
+"""
+
+TEST_KEY_PKCS8 = """
+-----BEGIN PRIVATE KEY-----
+MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCgshr1g7WzZ/iF
+tXbL+R2tTkzOsKVWeLli7Xe1+5W9jz+mEd/zWl/AD/QuM0ncNNkOQfyCVKgtzX4p
+k2yerf4QtdbwvVP36aqs6D0wH1AqkNfC338bbN1eNPNtZ2sG8muAppv4hjUBT7ck
+p3hrQ9VU596PFloYlM6M91mkIWGX76oig+pPSL62P82OwYbM3dBKIJsQaAEWbqf6
+V08D4MOUubt8zcS3FIpztLKQ+NPrzWGoPQoaHx0cruDZLWu2RdUTav4S3G0BLJFh
+TzekF6KGIUC/CdB6uOKFGobJxr9Pz/ZhiM/3PE3KBZXWeNS/+lerV2JB3MAAL2zc
+oPTHbUyXAgMBAAECggEADSAETVmeItKoUlIhgK8kCjkzKHigxdYjcjM/dq40q0sY
+IZoj0WO9dARO5zfKMxcKmXxbQaaPe/o6RpljeExxdsTwvYYBOzurHzydbsudR8oV
+IrYQKC8f9Odcm0Akq5MWTmeNUA200PAatH7h5NpMAjIpaUTOmWO3a9JyJb086Y9e
+Tv6Em0iNjZiyfuTz8q3BRcuZvwdXN3ZJVgTqJTN4iUfKmSbhDieIKHw1NieYbimD
+/LNyt6OMcZiU5RKvujY+kGLJBhJ5DO3/yl7+DfHeQjHrnyrddB2O8IwqBZp6wcmS
+5zTGX3vqclMsZCv2mKoVdWA3QlbDc36N7wumr9nYPQKBgQDbNsFGAaQ26bEsp37H
+ixegd9vhqeO8qFuTDMqNiKVs6t0ZV8PUmI0pB8ZzXxtWRiolUXqxZiqvilzhRz4m
+B/1lYKmljpSJdL4fX+t65MNW4pY7S/bqd/5Rt4j+DA77Ez9lcbVH9LwpWrhMXFvU
++qyFVglNq7vrDN82FHkrDVf3dQKBgQC7qXWE2hDMZ7CerNQu06GSLehQ65YBtRg/
+oQ62dcpEv7vxVn24fzwtMBFlmt3WY+N9Vr8O9dxuiX7PmBOIf1P3BW+qLU+bVExc
+AQychjKoHRsimkqDaX8XqjJkA7fD56zahLKLShKwmUdWoNvM9Vf8jokO7Ky6/Ckk
+8DR1YRI+WwKBgCJ3OWwIVglMNM8Qhp5ZAB36uleAno3K6pr0J+SY12mIg7q4UdHb
+KlaJImqBSYsG/Rlg9iluJsfyuSeyS9Mu7mhCCmo9QJ1b/YhxTFTGIvBGL9ii2O+E
+bjfq418OBkzzecQ5XIczs2HNIazf2JBj2TwditjrZ2BNZcra7TIXpv0BAoGAFhyP
+dknjdNI641dl4OaawQibuP0pPaZC5yxo+WLDpHobAQK56YVEZ+akBh7LSLmYUmpo
+NLEPeg7FXFyE4K5wDEkdsSWJnbm/DZAqFj9RtLFJQzh+0yEZEobBTTuK6ZBgs5az
+GuptbVWCLFjjzTCTX3TD/Xgxmg6gInQTGkmHxsUCgYB0p7w195Q+/fUhdCSnvwNQ
+VlrX5EBjvjkPZFQkXNCFoeIhQuADyQQJ7P9HHlrxSjHsbZOxTTcVQatm0kk8s6Et
+LAxPD9wtwVxzZHHVXCK+l8JNfBmiW14n6BQvkTlGgt1HFojrx4ubz86VwgP8FMuW
+Qq9ELI2S82qfw0VlsoZWxQ==
+-----END PRIVATE KEY-----
+"""
+
+
+
+def _b64url_decode(s):
+    return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))
+
+
+def _emsa_sha256(message, em_len):
+    prefix = bytes.fromhex("3031300d060960864801650304020105000420")
+    t = prefix + hashlib.sha256(message).digest()
+    return b"\x00\x01" + b"\xff" * (em_len - len(t) - 3) + b"\x00" + t
+
+
+def _rsa_verify(pem, message, signature):
+    """纯 Python RSA 公钥验签（与签名实现独立的验算）。"""
+    n, e, _d = _parse_rsa_private_key(pem)
+    em_len = (n.bit_length() + 7) // 8
+    m = pow(int.from_bytes(signature, "big"), e, n)
+    return m.to_bytes(em_len, "big") == _emsa_sha256(message, em_len)
+
+
+class Rs256TestCase(unittest.TestCase):
+    def test_sign_and_verify_pkcs1(self):
+        message = b"github app jwt signing input"
+        sig = rsa_sign_sha256(TEST_KEY_PKCS1, message)
+        n, _e, _d = _parse_rsa_private_key(TEST_KEY_PKCS1)
+        self.assertEqual(len(sig), (n.bit_length() + 7) // 8)
+        self.assertTrue(_rsa_verify(TEST_KEY_PKCS1, message, sig))
+
+    def test_sign_and_verify_pkcs8(self):
+        message = b"another signing input"
+        sig = rsa_sign_sha256(TEST_KEY_PKCS8, message)
+        self.assertTrue(_rsa_verify(TEST_KEY_PKCS8, message, sig))
+        self.assertFalse(_rsa_verify(TEST_KEY_PKCS8, b"tampered", sig))
+
+    def test_bad_key(self):
+        with self.assertRaises(AppError):
+            rsa_sign_sha256("not a key", b"x")
+        with self.assertRaises(AppError):
+            rsa_sign_sha256("-----BEGIN FOO-----\nAAAA\n-----END FOO-----",
+                            b"x")
+
+
+class JwtTestCase(unittest.TestCase):
+    def test_mint_structure_and_signature(self):
+        jwt = mint_app_jwt("5476837", TEST_KEY_PKCS1, now=1000000)
+        parts = jwt.split(".")
+        self.assertEqual(len(parts), 3)
+        header = json.loads(_b64url_decode(parts[0]))
+        self.assertEqual(header["alg"], "RS256")
+        payload = json.loads(_b64url_decode(parts[1]))
+        self.assertEqual(payload["iss"], "5476837")
+        self.assertEqual(payload["iat"], 1000000 - 60)
+        self.assertEqual(payload["exp"], 1000000 + 540)
+        signing_input = (parts[0] + "." + parts[1]).encode("ascii")
+        self.assertTrue(_rsa_verify(
+            TEST_KEY_PKCS1, signing_input, _b64url_decode(parts[2])))
+
+
+def _fake_exchange(expected_installation="175380082"):
+    def fake(method, url, headers, body):
+        assert method == "POST", method
+        assert "/app/installations/%s/access_tokens" % expected_installation in url, url
+        assert headers["Authorization"].startswith("Bearer "), headers
+        assert headers["Authorization"].count(".") == 2
+        return 201, {"token": "fake-installation-token",
+                     "expires_at": "2030-01-01T00:00:00Z",
+                     "permissions": {"contents": "write"},
+                     "repository_selection": "selected"}
+    return fake
+
+
+class TokenExchangeTestCase(unittest.TestCase):
+    def test_exchange_ok(self):
+        result = exchange_installation_token(
+            "5476837", TEST_KEY_PKCS1, "175380082",
+            http_client=_fake_exchange())
+        self.assertEqual(result["token"], "fake-installation-token")
+        self.assertEqual(result["repository_selection"], "selected")
+
+    def test_exchange_failure_raises(self):
+        def fake(method, url, headers, body):
+            return 401, {"message": "Bad credentials"}
+        with self.assertRaises(AppError) as ctx:
+            exchange_installation_token(
+                "5476837", TEST_KEY_PKCS1, "175380082", http_client=fake)
+        self.assertEqual(ctx.exception.reason, "token_exchange_failed")
+
+
+class ResolveTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.conn = _db_connect(self.db_path)
+
+    def tearDown(self):
+        self.conn.close()
+        self.tmp.cleanup()
+
+    def _save_config(self):
+        save_app_config(self.conn, {
+            "app_id": 5476837, "slug": "fake-app", "client_id": "Iv1.x",
+            "webhook_secret": "fake-secret", "pem": TEST_KEY_PKCS1,
+            "html_url": "https://github.com/apps/fake-app"})
+
+    def test_resolve_full_chain(self):
+        self._save_config()
+        record_installation_repos(self.conn, 175380082, "octo",
+                                  ["octo/a"])
+        token = resolve_app_token(self.conn, "octo/a",
+                                  http_client=_fake_exchange())
+        self.assertEqual(token, "fake-installation-token")
+
+    def test_resolve_falls_back_to_none(self):
+        # 无配置、无安装、换 token 失败：一律 None（调用方走 PAT）。
+        self.assertIsNone(resolve_app_token(
+            self.conn, "octo/a", http_client=_fake_exchange()))
+        self._save_config()
+        self.assertIsNone(resolve_app_token(
+            self.conn, "octo/a", http_client=_fake_exchange()))
+        record_installation_repos(self.conn, 175380082, "octo",
+                                  ["octo/a"])
+
+        def failing(method, url, headers, body):
+            return 401, {"message": "Bad credentials"}
+        self.assertIsNone(resolve_app_token(
+            self.conn, "octo/a", http_client=failing))
+
+    def test_find_installation(self):
+        record_installation_repos(self.conn, 11, "octo", ["octo/a"])
+        self.assertEqual(find_installation_for_repo(self.conn, "octo/a"),
+                         "11")
+        self.assertIsNone(find_installation_for_repo(self.conn, "octo/b"))
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_pr_app_token.py

@@ -0,0 +1,176 @@
+"""pr.py 的 App-token 路径单测：push 走 credential helper、建 PR
+走 GitHub API、gh 完全不被调用；remote 解析与失败分支。
+
+不联网、不跑真实 git/gh：git 用假 runner，GitHub API 用假
+客户端；token 值绝不出现在记录的命令里（只出现变量名）。
+"""
+
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.pr import (
+        PrStepError,
+        check_pr_ready,
+        create_task_pr,
+        parse_git_remote_owner_repo,
+    )
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.pr import (
+        PrStepError,
+        check_pr_ready,
+        create_task_pr,
+        parse_git_remote_owner_repo,
+    )
+
+
+def _result(returncode=0, stdout="", stderr=""):
+    return SimpleNamespace(returncode=returncode, stdout=stdout,
+                           stderr=stderr)
+
+
+class _Git:
+    """假 git runner：remote/branch/add/commit/push 全按脚本回。"""
+
+    def __init__(self, remote="https://github.com/octo/a.git"):
+        self.calls = []
+        self.remote = remote
+
+    def __call__(self, cmd, cwd):
+        self.calls.append(list(cmd))
+        if "remote" in cmd:
+            return _result(0, self.remote + "\n")
+        if "branch" in cmd:
+            return _result(0, "host/t1\n")
+        if "add" in cmd:
+            return _result(0, "")
+        if "commit" in cmd:
+            return _result(1, "",
+                           "nothing to commit, working tree clean")
+        if "push" in cmd:
+            return _result(0, "")
+        return _result(0, "")
+
+
+def _gh_forbidden(cmd, cwd):
+    raise AssertionError("App 路径不应调用 gh：%r" % (cmd,))
+
+
+class _FakeGithubApi:
+    """假 GitHub API：记录调用，GET 仓库回 default_branch，
+    POST pulls 回 PR 链接；失败模式可配。"""
+
+    def __init__(self, default_branch="main", pulls_status=201):
+        self.calls = []
+        self.default_branch = default_branch
+        self.pulls_status = pulls_status
+
+    def __call__(self, method, url, token, payload):
+        self.calls.append((method, url, token, payload))
+        if method == "GET" and url.endswith("/repos/octo/a"):
+            return 200, {"default_branch": self.default_branch}
+        if method == "POST" and url.endswith("/repos/octo/a/pulls"):
+            if self.pulls_status != 201:
+                return self.pulls_status, {"message": "Validation Failed"}
+            return 201, {"html_url": "https://github.com/octo/a/pull/42",
+                         "number": 42}
+        raise AssertionError("unexpected API call: %s %s" % (method, url))
+
+
+def _worktree(tmp):
+    d = os.path.join(tmp, "t1")
+    os.makedirs(d, exist_ok=True)
+    return d
+
+
+class ParseRemoteTestCase(unittest.TestCase):
+    def test_variants(self):
+        self.assertEqual(
+            parse_git_remote_owner_repo(
+                "https://github.com/octo/a.git"), ("octo", "a"))
+        self.assertEqual(
+            parse_git_remote_owner_repo(
+                "https://github.com/octo/a"), ("octo", "a"))
+        self.assertEqual(
+            parse_git_remote_owner_repo(
+                "https://user@github.com/octo/a.git"), ("octo", "a"))
+        self.assertEqual(
+            parse_git_remote_owner_repo(
+                "git@github.com:octo/a.git"), ("octo", "a"))
+        self.assertEqual(
+            parse_git_remote_owner_repo(
+                "ssh://git@github.com/octo/a.git"), ("octo", "a"))
+
+    def test_non_github_is_none(self):
+        self.assertIsNone(parse_git_remote_owner_repo(
+            "https://example.com/octo/a.git"))
+        self.assertIsNone(parse_git_remote_owner_repo(""))
+        self.assertIsNone(parse_git_remote_owner_repo(None))
+
+
+class AppPathTestCase(unittest.TestCase):
+    def test_push_uses_credential_helper_and_pr_via_api(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            _worktree(tmp)
+            git = _Git()
+            api = _FakeGithubApi()
+            result = create_task_pr(
+                "t1", 1, worktrees_root=tmp, git_runner=git,
+                gh_runner=_gh_forbidden, gh_which="/nonexistent/gh",
+                app_token="fake-app-token", github_api=api)
+            self.assertEqual(
+                result["pr_url"], "https://github.com/octo/a/pull/42")
+            self.assertEqual(result["pr_number"], 42)
+            self.assertEqual(result["branch"], "host/t1")
+            pushes = [c for c in git.calls if "push" in c]
+            self.assertEqual(len(pushes), 1)
+            helper = [a for a in pushes[0]
+                      if a.startswith("credential.helper=")][0]
+            self.assertIn("$GIT_TOKEN", helper)
+            self.assertNotIn("fake-app-token", " ".join(pushes[0]))
+            posts = [c for c in api.calls if c[0] == "POST"]
+            self.assertEqual(len(posts), 1)
+            self.assertEqual(posts[0][2], "fake-app-token")
+            self.assertEqual(posts[0][3]["head"], "host/t1")
+            self.assertEqual(posts[0][3]["base"], "main")
+            self.assertEqual(posts[0][3]["title"], "cloud: t1")
+
+    def test_app_path_skips_gh_checks(self):
+        # gh_which 指到不存在的路径、gh_runner 会炸：App 路径下
+        # check_pr_ready 仍应通过（只查 worktree + remote）。
+        with tempfile.TemporaryDirectory() as tmp:
+            _worktree(tmp)
+            self.assertTrue(check_pr_ready(
+                os.path.join(tmp, "t1"), git_runner=_Git(),
+                gh_which="/nonexistent/gh", gh_runner=_gh_forbidden,
+                app_token="x"))
+
+    def test_non_github_remote_fails_at_push(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            _worktree(tmp)
+            with self.assertRaises(PrStepError) as ctx:
+                create_task_pr(
+                    "t1", 1, worktrees_root=tmp,
+                    git_runner=_Git("https://example.com/octo/a.git"),
+                    gh_runner=_gh_forbidden, app_token="x",
+                    github_api=_FakeGithubApi())
+            self.assertEqual(ctx.exception.step, "push")
+
+    def test_api_failure_is_pr_create_step(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            _worktree(tmp)
+            with self.assertRaises(PrStepError) as ctx:
+                create_task_pr(
+                    "t1", 1, worktrees_root=tmp, git_runner=_Git(),
+                    gh_runner=_gh_forbidden, app_token="x",
+                    github_api=_FakeGithubApi(pulls_status=422))
+            self.assertEqual(ctx.exception.step, "pr_create")
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_server_pr_app_token.py

@@ -0,0 +1,248 @@
+"""POST /tasks/{id}/pr 的 App-token 接线单测（端点级）：
+
+App 路径：项目 repo_url 解析出仓库名、库里有该仓库的 App
+安装记录 -> 现换 installation token（fake 换 token 客户端 +
+测试专用 pem，JWT 纯标准库签发、不联网）-> push 走
+credential helper、建 PR 走 GitHub API（fake），响应里
+auth_via=app。
+PAT 回落：没有安装记录 -> 走原来的 gh 路径，auth_via=pat。
+
+git/GitHub API 一律 fake，不跑真实命令；token 值绝不出现
+在 git 命令记录里。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.db import connect as _db_connect
+    from cloud.github_app import (
+        record_installation_repos,
+        save_app_config,
+    )
+    from cloud.server import create_server
+    from cloud.tests.test_app_token import TEST_KEY_PKCS1
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.db import connect as _db_connect
+    from cloud.github_app import (
+        record_installation_repos,
+        save_app_config,
+    )
+    from cloud.server import create_server
+    from cloud.tests.test_app_token import TEST_KEY_PKCS1
+
+
+def _result(returncode=0, stdout="", stderr=""):
+    return SimpleNamespace(returncode=returncode, stdout=stdout,
+                           stderr=stderr)
+
+
+class _Git:
+    """假 git：remote/branch/add/commit/push，记录命令。"""
+
+    def __init__(self, remote, branch):
+        self.calls = []
+        self.remote = remote
+        self.branch = branch
+
+    def __call__(self, cmd, cwd):
+        self.calls.append(list(cmd))
+        if "remote" in cmd:
+            return _result(0, self.remote + "\n")
+        if "branch" in cmd:
+            return _result(0, self.branch + "\n")
+        if "add" in cmd:
+            return _result(0, "")
+        if "commit" in cmd:
+            return _result(1, "",
+                           "nothing to commit, working tree clean")
+        if "push" in cmd:
+            return _result(0, "")
+        return _result(0, "")
+
+
+class _Gh:
+    """假 gh：auth 通过、pr create 回链接。"""
+
+    def __init__(self, pr_url):
+        self.calls = []
+        self.pr_url = pr_url
+
+    def __call__(self, cmd, cwd):
+        self.calls.append(list(cmd))
+        if cmd[:2] == ["gh", "auth"]:
+            return _result(0, "", "")
+        if cmd[:2] == ["gh", "pr"]:
+            return _result(0, self.pr_url + "\n")
+        return _result(0, "")
+
+
+class _Api:
+    """假 GitHub API：GET 仓库回 default_branch，POST pulls 回 PR。"""
+
+    def __init__(self):
+        self.calls = []
+
+    def __call__(self, method, url, token, payload):
+        self.calls.append((method, url, token, payload))
+        if method == "GET":
+            return 200, {"default_branch": "main"}
+        if method == "POST":
+            return 201, {"html_url": "https://github.com/octo/a/pull/77",
+                         "number": 77}
+        raise AssertionError("unexpected API call: %s %s" % (method, url))
+
+
+def _token_client(method, url, headers, body):
+    """假换 token：JWT 已在前面签好（只验格式），直接回 token。"""
+    assert method == "POST", method
+    assert "/access_tokens" in url, url
+    assert headers["Authorization"].startswith("Bearer "), headers
+    return 201, {"token": "srv-app-token",
+                 "expires_at": "2030-01-01T00:00:00Z",
+                 "permissions": {"contents": "write"},
+                 "repository_selection": "selected"}
+
+
+class _Case(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.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()
+        self.worktrees = os.path.join(self.tmp.name, "worktrees")
+        os.makedirs(self.worktrees, exist_ok=True)
+        self.server.host_worktrees_dir = self.worktrees
+        self.git = None
+        self.api = _Api()
+        self.server.pr_github_api = self.api
+        self.server.pr_app_token_client = _token_client
+
+    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 _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection(
+            "127.0.0.1", self.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 _project(self, name, repo_url):
+        status, project = self._req(
+            "POST", "/projects",
+            {"name": name, "repo_url": repo_url, "auto_pr": 1},
+            self.owner_key)
+        self.assertEqual(status, 201, project)
+        return project["id"]
+
+    def _task(self, task_id, project_id):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, prompt, status,"
+                " depends_on, priority, created_by, created_at,"
+                " updated_at, last_heartbeat) VALUES(?, ?, ?, 'merged',"
+                " '[]', 0, 'system', 0, 0, NULL)",
+                (task_id, project_id, "do it"))
+            conn.commit()
+        finally:
+            conn.close()
+        os.makedirs(os.path.join(self.worktrees, task_id), exist_ok=True)
+
+    def test_app_path_via_endpoint(self):
+        project_id = self._project(
+            "octo/a", "https://github.com/octo/a")
+        conn = _db_connect(self.db_path)
+        try:
+            save_app_config(conn, {
+                "app_id": 5476837, "slug": "fake-app",
+                "client_id": "Iv1.x", "webhook_secret": "fake-secret",
+                "pem": TEST_KEY_PKCS1,
+                "html_url": "https://github.com/apps/fake-app"})
+            record_installation_repos(conn, 175380082, "octo",
+                                      ["octo/a"])
+        finally:
+            conn.close()
+        self.git = _Git("https://github.com/octo/a.git", "host/t-app-1")
+        self.server.pr_git_runner = self.git
+        self._task("t-app-1", project_id)
+        status, body = self._req(
+            "POST", "/tasks/t-app-1/pr", key=self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["pr_url"],
+                         "https://github.com/octo/a/pull/77")
+        self.assertEqual(body["pr_number"], 77)
+        self.assertEqual(body["auth_via"], "app")
+        pushes = [c for c in self.git.calls if "push" in c]
+        self.assertEqual(len(pushes), 1)
+        self.assertTrue(any(a.startswith("credential.helper=")
+                            for a in pushes[0]))
+        self.assertNotIn("srv-app-token",
+                         " ".join(" ".join(c) for c in self.git.calls))
+        posts = [c for c in self.api.calls if c[0] == "POST"]
+        self.assertEqual(len(posts), 1)
+        self.assertEqual(posts[0][2], "srv-app-token")
+
+    def test_pat_fallback_via_endpoint(self):
+        project_id = self._project(
+            "octo/b", "https://github.com/octo/b")
+        self.git = _Git("https://github.com/octo/b.git", "host/t-pat-1")
+        self.server.pr_git_runner = self.git
+        gh = _Gh("https://github.com/octo/b/pull/88")
+        self.server.pr_gh_runner = gh
+        self.server.pr_gh_which = "/usr/bin/gh"
+        self._task("t-pat-1", project_id)
+        status, body = self._req(
+            "POST", "/tasks/t-pat-1/pr", key=self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["pr_url"],
+                         "https://github.com/octo/b/pull/88")
+        self.assertEqual(body["auth_via"], "pat")
+        pushes = [c for c in self.git.calls if "push" in c]
+        self.assertEqual(len(pushes), 1)
+        self.assertFalse(any(a.startswith("credential.helper=")
+                              for a in pushes[0]))
+        self.assertEqual(self.api.calls, [])
+
+
+if __name__ == "__main__":
+    unittest.main()