diff --git a/cloud/server.py b/cloud/server.py index 31bd905..28bb811 100644 --- a/cloud/server.py +++ b/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," diff --git a/cloud/tests/test_longfunc_extra.py b/cloud/tests/test_longfunc_extra.py new file mode 100644 index 0000000..94b2b77 --- /dev/null +++ b/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()