refactor: extract minimal pure helpers from _continue_task, _task_wait, _task_thread_read and _metrics

_continue_task 226->168 via _validate_continue_options;
_task_wait 158->139 via _validate_wait_params;
_task_thread_read 187->152 via _summarize_trace_entries;
_metrics 123->104 via _summarize_metrics_tasks.
No behaviour, message, status or route change; error strings verbatim.
New tests: test_longfunc_extra (9). Cloud full suite: 1439 OK, skipped=4.
Follows #229 pattern; must merge after #229 and be rebased/checked then.

cloud/server.py

@@ -2046,6 +2046,186 @@ def _validate_task_goal_options(body, prompt, prompt_goal_target, plan):
             return (0, None, None, 0, "goal_max_rounds 超过上限 100 轮")
     return (0, goal, verify_command, goal_max_rounds, None)
 
+
+def _validate_continue_options(body, task, goal_target):
+    """_continue_task 的选项校验块（纯函数，原样平移）。
+
+    返回 (opts, error)：error 非空时调用方按原样发 400；
+    opts 含 model / effort / child_model / child_model_present /
+    child_effort / child_effort_present / approval / plan /
+    goal_mode，错误文案逐字不变。
+    """
+    model = body.get("model")
+    if model is not None:
+        if not isinstance(model, str):
+            return None, "model must be a string"
+        model = model.strip() or None
+    effort = body.get("effort")
+    if effort is not None and effort not in (
+            "low", "medium", "high", "xhigh", "max"):
+        return None, "effort must be low, medium, high, xhigh or max"
+    # 派生默认同款：child_model 空串 = 清空（跟随本任务
+    # model / effort），child_effort 空串 = 清空；不传沿用原值。
+    child_model = None
+    child_model_present = "child_model" in body
+    if child_model_present:
+        if body.get("child_model") is not None \
+                and not isinstance(body.get("child_model"), str):
+            return None, "child_model must be a string"
+        raw_child_model = body.get("child_model")
+        child_model = raw_child_model.strip() \
+            if isinstance(raw_child_model, str) else None
+        if child_model == "":
+            child_model = None
+    child_effort = None
+    child_effort_present = "child_effort" in body
+    if child_effort_present:
+        raw_child_effort = body.get("child_effort")
+        if raw_child_effort is None or raw_child_effort == "":
+            child_effort = None
+        elif raw_child_effort not in (
+                "low", "medium", "high", "xhigh", "max"):
+            return None, ("child_effort must be low, medium, high,"
+                          " xhigh or max")
+        else:
+            child_effort = raw_child_effort
+    approval = body.get("approval")
+    if approval is not None and approval not in (
+            "ask", "auto", "full"):
+        return None, "approval must be ask, auto or full"
+    plan = None
+    if "plan" in body:
+        plan = 1 if body["plan"] in (True, 1) else 0
+    goal_mode = None
+    if "goal_mode" in body:
+        goal_mode = 1 if body["goal_mode"] in (True, 1) else 0
+    if goal_target is not None:
+        # /goal 命令本身就是显式 goal 请求，优先于 body 里的
+        # goal_mode=0（消息比选项更具体），plan 冲突仍在下面 400。
+        goal_mode = 1
+    plan_eff = plan if plan is not None else (task["plan"] or 0)
+    goal_eff = goal_mode if goal_mode is not None \
+        else (task["goal_mode"] or 0)
+    approval_eff = approval if approval is not None \
+        else task["approval"]
+    if plan_eff and approval_eff == "full":
+        return None, "plan 模式与 approval=full 冲突（只读 vs 完全访问）"
+    if plan_eff and goal_eff:
+        return None, "plan 模式与 goal 模式冲突"
+    return {
+        "model": model, "effort": effort,
+        "child_model": child_model,
+        "child_model_present": child_model_present,
+        "child_effort": child_effort,
+        "child_effort_present": child_effort_present,
+        "approval": approval, "plan": plan, "goal_mode": goal_mode,
+    }, None
+
+
+def _validate_wait_params(body):
+    """_task_wait 的参数校验块（纯函数，原样平移）。
+
+    返回 (timeout, requested, error_payload)：error_payload 非空
+    时调用方按原样发 400；requested 为 None 表示缺省（等全部
+    coordinator 子任务），其余错误文案与 count 键逐字不变。
+    """
+    timeout = body.get("timeout_seconds", 30)
+    if isinstance(timeout, bool) or not isinstance(
+            timeout, (int, float)):
+        return None, None, {"error": "timeout_seconds must be a number"}
+    if timeout < 0 or timeout > 120:
+        return None, None, {
+            "error": "timeout_seconds must be between 0 and 120"}
+    requested = body.get("task_ids")
+    if requested is not None and (
+            not isinstance(requested, list) or any(
+                not isinstance(x, str) for x in requested)):
+        return None, None, {"error": "task_ids must be a list"}
+    if requested is not None and len(requested) == 0:
+        return None, None, {"error": "task_ids must not be empty"}
+    if requested is not None and len(requested) > 8:
+        return None, None, {
+            "error": "at most 8 threads per wait",
+            "count": len(requested)}
+    return timeout, requested, None
+
+
+def _trace_activity(entry):
+    """_task_thread_read 的单条链路活动摘要（纯函数，原样平移）。"""
+    if not isinstance(entry, dict):
+        return {"kind": "unknown"}
+    if entry.get("kind") == "tool":
+        # tool 条目没有 text/command，旧代码只取 tool 名，
+        # 丢了 server 与 status；拼成可读的一行。
+        text = "%s/%s: %s" % (
+            entry.get("server") or "",
+            entry.get("tool") or "",
+            entry.get("status") or "")
+    else:
+        text = (entry.get("text") or entry.get("command")
+                or entry.get("path") or entry.get("summary")
+                or entry.get("tool") or "")
+    return {"kind": entry.get("kind"),
+            "id": entry.get("id"),
+            "text": str(text)[:300]}
+
+
+def _summarize_trace_entries(entries):
+    """_task_thread_read 的链路摘要块（纯函数，原样平移）。
+
+    返回 (changed_files, commands, recent_activity)：改动文件
+    去重取后 50、命令取后 20、最近动态取后 10，截断口径不变。
+    """
+    # 改动文件 / 命令 / 最近动态：只取链路里已有的事实，
+    # 去重与截断在服务端做，不回整条链路原文。
+    changed = {}
+    for entry in entries:
+        if isinstance(entry, dict) and entry.get("kind") == "file" \
+                and entry.get("path"):
+            changed[entry["path"]] = {
+                "path": entry["path"],
+                "change": entry.get("change") or "",
+                "status": entry.get("status") or ""}
+    changed_files = list(changed.values())[-50:]
+    commands = [{
+        "command": (e.get("command") or "")[:500],
+        "exit_code": e.get("exitCode"),
+        "duration_ms": e.get("durationMs"),
+    } for e in entries if isinstance(e, dict)
+        and e.get("kind") == "exec"][-20:]
+    recent_activity = [_trace_activity(e) for e in entries[-10:]]
+    return changed_files, commands, recent_activity
+
+
+def _summarize_metrics_tasks(rows, member_ids, now):
+    """_metrics 的任务行聚合块（纯函数，原样平移，两遍合一遍）。
+
+    返回 (by_status, durations, failures_last_24h,
+    running_over_300s)：按状态分计数（七个状态键恒存在）、
+    完成耗时、近 24h 失败数、跑超 300 秒的 running 数，
+    口径与原两遍循环逐字一致。
+    """
+    # 按状态分计数：七个状态键恒存在（缺失为 0），调用方
+    # 不用再猜某个状态是“零”还是“没统计”。
+    by_status = {key: 0 for key in TASK_STATUS_KEYS}
+    durations = []
+    failures_last_24h = 0
+    running_over_300s = 0
+    for row in rows:
+        if row["project_id"] not in member_ids:
+            continue
+        by_status[row["status"]] = by_status.get(row["status"], 0) + 1
+        if row["started_at"] and row["finished_at"]:
+            durations.append(row["finished_at"] - row["started_at"])
+        if row["status"] in ("failed", "merge-failed") \
+                and row["finished_at"] \
+                and now - row["finished_at"] <= 86400:
+            failures_last_24h += 1
+        if row["status"] == "running" and row["started_at"] \
+                and now - row["started_at"] > 300:
+            running_over_300s += 1
+    return by_status, durations, failures_last_24h, running_over_300s
+
 class CloudControlPlaneHandler(BaseHTTPRequestHandler):
     """控制面请求处理：除 /health 与 /webhooks/* 外一律要求 X-API-Key。"""
 
@@ -6439,43 +6619,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                                 " no thread")
                     else:
                         thread = full
-        # 改动文件 / 命令 / 最近动态：只取链路里已有的事实，
-        # 去重与截断在服务端做，不回整条链路原文。
-        changed = {}
-        for entry in entries:
-            if isinstance(entry, dict) and entry.get("kind") == "file" \
-                    and entry.get("path"):
-                changed[entry["path"]] = {
-                    "path": entry["path"],
-                    "change": entry.get("change") or "",
-                    "status": entry.get("status") or ""}
-        changed_files = list(changed.values())[-50:]
-        commands = [{
-            "command": (e.get("command") or "")[:500],
-            "exit_code": e.get("exitCode"),
-            "duration_ms": e.get("durationMs"),
-        } for e in entries if isinstance(e, dict)
-            and e.get("kind") == "exec"][-20:]
-
-        def _activity(entry):
-            if not isinstance(entry, dict):
-                return {"kind": "unknown"}
-            if entry.get("kind") == "tool":
-                # tool 条目没有 text/command，旧代码只取 tool 名，
-                # 丢了 server 与 status；拼成可读的一行。
-                text = "%s/%s: %s" % (
-                    entry.get("server") or "",
-                    entry.get("tool") or "",
-                    entry.get("status") or "")
-            else:
-                text = (entry.get("text") or entry.get("command")
-                        or entry.get("path") or entry.get("summary")
-                        or entry.get("tool") or "")
-            return {"kind": entry.get("kind"),
-                    "id": entry.get("id"),
-                    "text": str(text)[:300]}
-
-        recent_activity = [_activity(e) for e in entries[-10:]]
+        changed_files, commands, recent_activity = \
+            _summarize_trace_entries(entries)
         # 待审批与待处理后续消息（阻塞与“需不需要人”的来源）。
         conn = self._db()
         try:
@@ -7320,28 +7465,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         if err is not None:
             self._send_json(err[0], {"error": err[1]})
             return
-        timeout = body.get("timeout_seconds", 30)
-        if isinstance(timeout, bool) or not isinstance(
-                timeout, (int, float)):
-            self._send_json(400, {"error": "timeout_seconds must be a number"})
-            return
-        if timeout < 0 or timeout > 120:
-            self._send_json(400, {
-                "error": "timeout_seconds must be between 0 and 120"})
-            return
-        requested = body.get("task_ids")
-        if requested is not None and (
-                not isinstance(requested, list) or any(
-                    not isinstance(x, str) for x in requested)):
-            self._send_json(400, {"error": "task_ids must be a list"})
-            return
-        if requested is not None and len(requested) == 0:
-            self._send_json(400, {"error": "task_ids must not be empty"})
-            return
-        if requested is not None and len(requested) > 8:
-            self._send_json(400, {
-                "error": "at most 8 threads per wait",
-                "count": len(requested)})
+        timeout, requested, werr = _validate_wait_params(body)
+        if werr is not None:
+            self._send_json(400, werr)
             return
         conn = self._db()
         try:
@@ -7990,77 +8116,19 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             self._send_json(400, {"error": "goal 目标不能为空"})
             return
         resume_text = goal_target if goal_target else message
-        model = body.get("model")
-        if model is not None:
-            if not isinstance(model, str):
-                self._send_json(400, {
-                    "error": "model must be a string"})
-                return
-            model = model.strip() or None
-        effort = body.get("effort")
-        if effort is not None and effort not in (
-                "low", "medium", "high", "xhigh", "max"):
-            self._send_json(400, {
-                "error": "effort must be low, medium, high, xhigh or max"})
-            return
-        # 派生默认同款：child_model 空串 = 清空（跟随本任务
-        # model / effort），child_effort 空串 = 清空；不传沿用原值。
-        child_model = None
-        child_model_present = "child_model" in body
-        if child_model_present:
-            if body.get("child_model") is not None \
-                    and not isinstance(body.get("child_model"), str):
-                self._send_json(400, {
-                    "error": "child_model must be a string"})
-                return
-            raw_child_model = body.get("child_model")
-            child_model = raw_child_model.strip() \
-                if isinstance(raw_child_model, str) else None
-            if child_model == "":
-                child_model = None
-        child_effort = None
-        child_effort_present = "child_effort" in body
-        if child_effort_present:
-            raw_child_effort = body.get("child_effort")
-            if raw_child_effort is None or raw_child_effort == "":
-                child_effort = None
-            elif raw_child_effort not in (
-                    "low", "medium", "high", "xhigh", "max"):
-                self._send_json(400, {
-                    "error": "child_effort must be low, medium, high,"
-                             " xhigh or max"})
-                return
-            else:
-                child_effort = raw_child_effort
-        approval = body.get("approval")
-        if approval is not None and approval not in (
-                "ask", "auto", "full"):
-            self._send_json(400, {
-                "error": "approval must be ask, auto or full"})
-            return
-        plan = None
-        if "plan" in body:
-            plan = 1 if body["plan"] in (True, 1) else 0
-        goal_mode = None
-        if "goal_mode" in body:
-            goal_mode = 1 if body["goal_mode"] in (True, 1) else 0
-        if goal_target is not None:
-            # /goal 命令本身就是显式 goal 请求，优先于 body 里的
-            # goal_mode=0（消息比选项更具体），plan 冲突仍在下面 400。
-            goal_mode = 1
-        plan_eff = plan if plan is not None else (task["plan"] or 0)
-        goal_eff = goal_mode if goal_mode is not None \
-            else (task["goal_mode"] or 0)
-        approval_eff = approval if approval is not None \
-            else task["approval"]
-        if plan_eff and approval_eff == "full":
-            self._send_json(400, {
-                "error": "plan 模式与 approval=full 冲突（只读 vs 完全访问）"})
-            return
-        if plan_eff and goal_eff:
-            self._send_json(400, {
-                "error": "plan 模式与 goal 模式冲突"})
+        opts, cerr = _validate_continue_options(body, task, goal_target)
+        if cerr:
+            self._send_json(400, {"error": cerr})
             return
+        model = opts["model"]
+        effort = opts["effort"]
+        child_model = opts["child_model"]
+        child_model_present = opts["child_model_present"]
+        child_effort = opts["child_effort"]
+        child_effort_present = opts["child_effort_present"]
+        approval = opts["approval"]
+        plan = opts["plan"]
+        goal_mode = opts["goal_mode"]
         sets = ["status = 'queued'", "resume_message = ?",
                 "result_text = NULL", "error_text = NULL",
                 "actual_model = NULL", "updated_at = ?"]
@@ -8993,35 +9061,16 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 "SELECT project_id, status, started_at, finished_at"
                 " FROM tasks"
             ).fetchall()
-            # 按状态分计数：七个状态键恒存在（缺失为 0），调用方
-            # 不用再猜某个状态是“零”还是“没统计”。
-            by_status = {key: 0 for key in TASK_STATUS_KEYS}
-            durations = []
-            for row in rows:
-                if row["project_id"] not in member_ids:
-                    continue
-                by_status[row["status"]] = by_status.get(row["status"], 0) + 1
-                if row["started_at"] and row["finished_at"]:
-                    durations.append(row["finished_at"] - row["started_at"])
+            # 新增可观测字段（只加键，旧键语义不动）：队列深度、
+            # 近 24h 失败数、跑超 300 秒的 running 任务数。
+            now = time.time()
+            (by_status, durations, failures_last_24h,
+             running_over_300s) = _summarize_metrics_tasks(
+                 rows, member_ids, now)
             total = sum(by_status.values())
             done = by_status.get("completed", 0) + by_status.get("merged", 0)
             failed = by_status.get("failed", 0) + by_status.get("merge-failed", 0)
             finished = done + failed
-            # 新增可观测字段（只加键，旧键语义不动）：队列深度、
-            # 近 24h 失败数、跑超 300 秒的 running 任务数。
-            now = time.time()
-            failures_last_24h = 0
-            running_over_300s = 0
-            for row in rows:
-                if row["project_id"] not in member_ids:
-                    continue
-                if row["status"] in ("failed", "merge-failed") \
-                        and row["finished_at"] \
-                        and now - row["finished_at"] <= 86400:
-                    failures_last_24h += 1
-                if row["status"] == "running" and row["started_at"] \
-                        and now - row["started_at"] > 300:
-                    running_over_300s += 1
             try:
                 cost_rows = conn.execute(
                     "SELECT project_id, model, source, total_tokens,"

cloud/tests/test_longfunc_extra.py

@@ -0,0 +1,177 @@
+"""长函数小提取第二轨（#229 之后）：四处纯函数钉住测试。
+
+只钉住本轨抽出的四个纯函数（成功路径与错误串），不改原有
+断言语义；原路由行为不变由 test_task_continue /
+test_thread_wait_ops / test_thread_ops 等定向套件覆盖。
+"""
+
+import sys
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.server import (
+        TASK_STATUS_KEYS,
+        _summarize_metrics_tasks,
+        _summarize_trace_entries,
+        _validate_continue_options,
+        _validate_wait_params,
+    )
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.server import (
+        TASK_STATUS_KEYS,
+        _summarize_metrics_tasks,
+        _summarize_trace_entries,
+        _validate_continue_options,
+        _validate_wait_params,
+    )
+
+
+def _task(**over):
+    task = {"plan": 0, "goal_mode": 0, "approval": "ask"}
+    task.update(over)
+    return task
+
+
+class ValidateContinueOptionsTestCase(unittest.TestCase):
+    def test_success_defaults_from_task(self):
+        opts, err = _validate_continue_options({}, _task(), None)
+        self.assertIsNone(err)
+        self.assertIsNone(opts["model"])
+        self.assertIsNone(opts["effort"])
+        self.assertFalse(opts["child_model_present"])
+        self.assertFalse(opts["child_effort_present"])
+        self.assertIsNone(opts["approval"])
+        self.assertIsNone(opts["plan"])
+        self.assertIsNone(opts["goal_mode"])
+
+    def test_success_explicit_and_empty_child_clears(self):
+        opts, err = _validate_continue_options(
+            {"model": "  m  ", "effort": "low", "child_model": "",
+             "child_effort": "", "approval": "auto", "plan": True,
+             "goal_mode": 0},
+            _task(), None)
+        self.assertIsNone(err)
+        self.assertEqual(opts["model"], "m")
+        self.assertEqual(opts["effort"], "low")
+        self.assertIsNone(opts["child_model"])
+        self.assertTrue(opts["child_model_present"])
+        self.assertIsNone(opts["child_effort"])
+        self.assertTrue(opts["child_effort_present"])
+        self.assertEqual(opts["plan"], 1)
+        self.assertEqual(opts["goal_mode"], 0)
+
+    def test_goal_command_forces_goal_mode(self):
+        opts, err = _validate_continue_options(
+            {"goal_mode": 0}, _task(), "do X")
+        self.assertIsNone(err)
+        self.assertEqual(opts["goal_mode"], 1)
+
+    def test_error_strings_verbatim(self):
+        cases = [
+            ({"model": 1}, "model must be a string"),
+            ({"effort": "huge"},
+             "effort must be low, medium, high, xhigh or max"),
+            ({"child_model": 1}, "child_model must be a string"),
+            ({"child_effort": "huge"},
+             "child_effort must be low, medium, high, xhigh or max"),
+            ({"approval": "all"}, "approval must be ask, auto or full"),
+            ({"plan": True, "approval": "full"},
+             "plan 模式与 approval=full 冲突（只读 vs 完全访问）"),
+            ({"plan": True, "goal_mode": 1}, "plan 模式与 goal 模式冲突"),
+        ]
+        for body, want in cases:
+            with self.subTest(body=body):
+                opts, err = _validate_continue_options(body, _task(), None)
+                self.assertIsNone(opts)
+                self.assertEqual(err, want)
+
+
+class ValidateWaitParamsTestCase(unittest.TestCase):
+    def test_default_and_valid(self):
+        timeout, requested, err = _validate_wait_params({})
+        self.assertEqual(timeout, 30)
+        self.assertIsNone(requested)
+        self.assertIsNone(err)
+        timeout, requested, err = _validate_wait_params(
+            {"timeout_seconds": 5, "task_ids": ["a"]})
+        self.assertEqual((timeout, requested, err), (5, ["a"], None))
+
+    def test_error_payloads_verbatim(self):
+        cases = [
+            ({"timeout_seconds": True},
+             {"error": "timeout_seconds must be a number"}),
+            ({"timeout_seconds": 121},
+             {"error": "timeout_seconds must be between 0 and 120"}),
+            ({"task_ids": "a"}, {"error": "task_ids must be a list"}),
+            ({"task_ids": [1]}, {"error": "task_ids must be a list"}),
+            ({"task_ids": []}, {"error": "task_ids must not be empty"}),
+            ({"task_ids": ["a"] * 9},
+             {"error": "at most 8 threads per wait", "count": 9}),
+        ]
+        for body, want in cases:
+            with self.subTest(body=body):
+                _, _, err = _validate_wait_params(body)
+                self.assertEqual(err, want)
+
+
+class SummarizeTraceEntriesTestCase(unittest.TestCase):
+    def test_changed_commands_activity(self):
+        entries = [
+            {"kind": "file", "path": "a.py", "change": "mod",
+             "status": "ok"},
+            {"kind": "file", "path": "a.py", "change": "mod2",
+             "status": "ok2"},
+            {"kind": "exec", "command": "pytest", "exitCode": 0,
+             "durationMs": 7},
+            {"kind": "tool", "server": "s", "tool": "t",
+             "status": "done", "id": "x"},
+            {"kind": "message", "text": "hi", "id": "y"},
+            "not-a-dict",
+        ]
+        changed, commands, activity = _summarize_trace_entries(entries)
+        self.assertEqual(changed, [{"path": "a.py", "change": "mod2",
+                                    "status": "ok2"}])
+        self.assertEqual(commands, [{"command": "pytest", "exit_code": 0,
+                                     "duration_ms": 7}])
+        self.assertEqual(activity[3],
+                         {"kind": "tool", "id": "x", "text": "s/t: done"})
+        self.assertEqual(activity[4],
+                         {"kind": "message", "id": "y", "text": "hi"})
+        self.assertEqual(activity[5], {"kind": "unknown"})
+
+    def test_truncation_windows(self):
+        entries = [{"kind": "message", "text": "t%d" % i, "id": i}
+                   for i in range(15)]
+        _, _, activity = _summarize_trace_entries(entries)
+        self.assertEqual(len(activity), 10)
+        self.assertEqual(activity[0]["text"], "t5")
+
+
+class SummarizeMetricsTasksTestCase(unittest.TestCase):
+    def test_aggregation_and_member_filter(self):
+        now = 1000000.0
+        rows = [
+            {"project_id": "p1", "status": "completed",
+             "started_at": 100.0, "finished_at": 160.0},
+            {"project_id": "p1", "status": "failed",
+             "started_at": None, "finished_at": now - 100},
+            {"project_id": "p1", "status": "running",
+             "started_at": now - 400, "finished_at": None},
+            {"project_id": "p2", "status": "completed",
+             "started_at": 0.0, "finished_at": 10.0},
+        ]
+        by_status, durations, failures, over = _summarize_metrics_tasks(
+            rows, {"p1"}, now)
+        self.assertEqual(set(by_status), set(TASK_STATUS_KEYS))
+        self.assertEqual(by_status["completed"], 1)
+        self.assertEqual(by_status["failed"], 1)
+        self.assertEqual(by_status["running"], 1)
+        self.assertEqual(durations, [60.0])
+        self.assertEqual(failures, 1)
+        self.assertEqual(over, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()