Merge pull request #41 from openqj/pr-timeout

pr 超时：git/gh 子进程默认 120 秒超时

cloud/README.md

@@ -849,7 +849,11 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳
   （先读仓库 default_branch 当 base，不猜 main）；没有安装
   记录或换 token 失败时自动回落原来的 gh + PAT 路径
   （auth_via=pat），不因此报错；App 没装的仓库与 gh 路径
-  的行为与原来完全一致。push 或 pr create
+  的行为与原来完全一致。子进程超时：push、pr create、
+  commit、remote 查询、gh 检查全部走带超时的子进程调用
+  （默认 120 秒，`CLOUD_PR_SUBPROCESS_TIMEOUT` 可改）；
+  远端通路卡死时 handler 线程在时限内结束、返回 502，
+  任务保持 merged、可安全重调，不再永久挂住。push 或 pr create
   失败时返回 502 且错误文本写清是哪一步（commit、push 还是
   pr_create），
   任务状态保持 merged 不变，绝不把任务标 failed。顺序：任务

cloud/pr.py

@@ -47,6 +47,19 @@ host/<host 任务名>，worktree 在 <仓库根>/work/host-worktrees/
   （先读仓库 default_branch 再建，base 不猜 main）。没传
   app_token 时走原来的 gh 路径，两条路不互相干扰；App 没装
   在该仓库时调用方不传 token、自动回落 PAT。
+
+  超时：本模块所有 git/gh 子进程都有超时（默认 120 秒，
+  环境变量 CLOUD_PR_SUBPROCESS_TIMEOUT 可改，非法值回落
+  默认）。此前没有超时：远端通路卡死时（主机到 GitHub 的
+  间歇性劣化曾让连接挂 120 秒以上）handler 线程会永久挂住，
+  客户端断开也收不回来。超时只保证 handler 线程在时限内
+  结束并返回 502（任务保持 merged、可安全重调），不能把
+  结果送给已经断开的客户端；超时发生在 push 时远端引用
+  未必已更新（git push 按引用原子更新，传一半被杀不会
+  留下半个分支），重调安全。超时落在就绪探针（remote
+  查询、gh auth）上时按原有跳过 reason 返回（no_remote /
+  gh_not_authenticated），不单独区分——本地 git 命令正常
+  在 1 秒内完成，探针超时只意味着 git 本身已 wedged。
 """
 
 import os
@@ -60,6 +73,26 @@ _SKIP_NO_REMOTE = "no_remote"
 _SKIP_NO_GH = "no_gh"
 _SKIP_NOT_AUTHED = "gh_not_authenticated"
 
+# 子进程超时（秒）：远端通路卡死时 handler 线程的上限。
+# 正常 git 本地操作 < 1 秒、健康远端 push 几秒，120 秒只在
+# 通路劣化时才触发（此前实测主机到 GitHub 的挂起超过 120 秒）。
+_DEFAULT_SUBPROCESS_TIMEOUT = 120
+
+
+def _pr_subprocess_timeout():
+    """超时取值：环境变量 CLOUD_PR_SUBPROCESS_TIMEOUT（整数秒），
+    未设、非整数或 <= 0 时回落默认。"""
+    raw = os.environ.get("CLOUD_PR_SUBPROCESS_TIMEOUT")
+    if raw is None:
+        return _DEFAULT_SUBPROCESS_TIMEOUT
+    try:
+        value = int(raw)
+    except (TypeError, ValueError):
+        return _DEFAULT_SUBPROCESS_TIMEOUT
+    if value <= 0:
+        return _DEFAULT_SUBPROCESS_TIMEOUT
+    return value
+
 
 class PrSkipped(Exception):
     """跳过：reason 见上列常量，调用方返回 200 + pr_skipped。"""
@@ -105,19 +138,32 @@ def _scrub_url(text):
     return re.sub(r"(https?://)[^/@\s]+@", r"\1", text)
 
 
-def _run(cmd, cwd=None, runner=None, env=None):
+def _run(cmd, cwd=None, runner=None, env=None, timeout=None):
     """命令唯一入口：runner 供单测注入，返回带 returncode/stdout/
     stderr 的对象；真实路径走 subprocess。env 只在真实路径生效
     （与进程环境合并），注入的 runner 不收 env——token 绝不进
-    假 runner 记录的命令，也不进真实路径的命令行参数。"""
+    假 runner 记录的命令，也不进真实路径的命令行参数。
+    超时：真实路径一律带 timeout（默认见 _pr_subprocess_timeout），
+    超时杀掉子进程并返回 returncode=124 的结果，调用方按原有
+    失败路径处理（push/pr_create/commit 抛 PrStepError，探针
+    按跳过处理）；注入的 runner 不收 timeout，单测用返回值
+    模拟超时映射，不模拟真实挂起。"""
     if runner is not None:
         return runner(cmd, cwd)
+    if timeout is None:
+        timeout = _pr_subprocess_timeout()
     full_env = None
     if env:
         full_env = dict(os.environ)
         full_env.update(env)
+    try:
         return subprocess.run(cmd, cwd=cwd, capture_output=True, text=True,
-                          env=full_env)
+                              env=full_env, timeout=timeout)
+    except subprocess.TimeoutExpired:
+        return subprocess.CompletedProcess(
+            cmd, 124, "",
+            "command timed out after %s seconds: %s"
+            % (timeout, " ".join(cmd)))
 
 
 _GITHUB_REMOTE_RE = re.compile(

cloud/tests/test_pr_timeout.py

@@ -0,0 +1,133 @@
+"""pr.py 子进程超时单测。
+
+真实路径：sleep 命令实测超时真的杀进程、真的按时返回
+（不是等命令跑完）；环境变量改超时、非法值回落默认。
+映射路径：假 runner 回 124 超时结果时，push/commit 按原有
+失败路径抛 PrStepError（step 对得上），不挂住、不误报成功。
+"""
+
+import os
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.pr import (
+        PrStepError,
+        _pr_subprocess_timeout,
+        _run,
+        create_task_pr,
+    )
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.pr import (
+        PrStepError,
+        _pr_subprocess_timeout,
+        _run,
+        create_task_pr,
+    )
+
+
+def _result(returncode=0, stdout="", stderr=""):
+    return SimpleNamespace(returncode=returncode, stdout=stdout,
+                           stderr=stderr)
+
+
+class _Git:
+    """假 git：push/commit 的失败模式可配，其余按脚本回。"""
+
+    def __init__(self, push="ok", commit="nothing"):
+        self.push = push
+        self.commit = commit
+
+    def __call__(self, cmd, cwd):
+        if "remote" in cmd:
+            return _result(0, "https://github.com/octo/a.git\n")
+        if "branch" in cmd:
+            return _result(0, "host/t1\n")
+        if "add" in cmd:
+            return _result(0, "")
+        if "commit" in cmd:
+            if self.commit == "timeout":
+                return _result(124, "", "command timed out")
+            return _result(1, "",
+                           "nothing to commit, working tree clean")
+        if "push" in cmd:
+            if self.push == "timeout":
+                return _result(124, "", "command timed out")
+            return _result(0, "")
+        return _result(0, "")
+
+
+def _gh(cmd, cwd):
+    if cmd[:2] == ["gh", "auth"]:
+        return _result(0, "", "")
+    if cmd[:2] == ["gh", "pr"]:
+        return _result(0, "https://github.com/octo/a/pull/9\n")
+    return _result(0, "")
+
+
+class RunTimeoutTestCase(unittest.TestCase):
+    def test_real_sleep_is_killed_on_timeout(self):
+        start = time.monotonic()
+        r = _run(["sleep", "3"], timeout=1)
+        elapsed = time.monotonic() - start
+        self.assertEqual(r.returncode, 124)
+        self.assertIn("timed out", r.stderr)
+        self.assertLess(elapsed, 2.5)
+
+    def test_fast_command_ok_with_default_timeout(self):
+        r = _run(["true"])
+        self.assertEqual(r.returncode, 0)
+
+    def test_env_override_and_invalid_fallback(self):
+        old = os.environ.get("CLOUD_PR_SUBPROCESS_TIMEOUT")
+        try:
+            os.environ["CLOUD_PR_SUBPROCESS_TIMEOUT"] = "1"
+            self.assertEqual(_pr_subprocess_timeout(), 1)
+            start = time.monotonic()
+            r = _run(["sleep", "3"])
+            self.assertEqual(r.returncode, 124)
+            self.assertLess(time.monotonic() - start, 2.5)
+            os.environ["CLOUD_PR_SUBPROCESS_TIMEOUT"] = "abc"
+            self.assertEqual(_pr_subprocess_timeout(), 120)
+            os.environ["CLOUD_PR_SUBPROCESS_TIMEOUT"] = "0"
+            self.assertEqual(_pr_subprocess_timeout(), 120)
+        finally:
+            if old is None:
+                os.environ.pop("CLOUD_PR_SUBPROCESS_TIMEOUT", None)
+            else:
+                os.environ["CLOUD_PR_SUBPROCESS_TIMEOUT"] = old
+
+
+class TimeoutMappingTestCase(unittest.TestCase):
+    def _worktree(self, tmp):
+        d = os.path.join(tmp, "t1")
+        os.makedirs(d, exist_ok=True)
+        return d
+
+    def test_push_timeout_is_pr_step_error(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            self._worktree(tmp)
+            with self.assertRaises(PrStepError) as ctx:
+                create_task_pr("t1", 1, worktrees_root=tmp,
+                               git_runner=_Git(push="timeout"),
+                               gh_runner=_gh, gh_which="/usr/bin/gh")
+            self.assertEqual(ctx.exception.step, "push")
+            self.assertIn("timed out", ctx.exception.message)
+
+    def test_commit_timeout_is_pr_step_error(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            self._worktree(tmp)
+            with self.assertRaises(PrStepError) as ctx:
+                create_task_pr("t1", 1, worktrees_root=tmp,
+                               git_runner=_Git(commit="timeout"),
+                               gh_runner=_gh, gh_which="/usr/bin/gh")
+            self.assertEqual(ctx.exception.step, "commit")
+
+
+if __name__ == "__main__":
+    unittest.main()