From 5ee9d291dd5436ef447fabea1629e2c5f2c64c64 Mon Sep 17 00:00:00 2001 From: host Date: Mon, 26 Oct 2026 18:11:12 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BA=91=E7=AB=AF=20goal/merge/review=20?= =?UTF-8?q?=E4=B8=89=E4=BA=8B=E4=BB=B6=EF=BC=9Agoal=20=E5=BE=AA=E7=8E=AF?= =?UTF-8?q?=20+=20PR=20=E4=BA=8B=E4=BB=B6=E9=80=9A=E7=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 此前 goal-unmet/merge-failed/review-requested 只有允许列表， 云端没有生产者（概念只在本地 host.py）。本轮给三者接真生产者： - goal：POST /tasks 接受 goal/verify_command/goal_max_rounds （迁移补列，SQLite 与 Postgres 两支）；带 verify_command 的 任务走 host.py goal 循环（Local/Docker 执行器 goal_task， turn→验测→喂回重试，缺省 3 轮）；达标 completed，未达标 failed 且只发 goal-unmet、不重复发 failed；goal 模式下 worker 级重试/fallback 关闭，同任务 /retry 会撞 host 任务名已存在 （已在文档写明，须新建任务）；CloudExecutor 无 goal 概念， 如实失败不降级 - review-requested：POST /tasks/{id}/pr 建 PR 成功时发 （payload 含 pr_url/pr_number） - merge-failed：PR 落盘（push/建 PR）失败 502 时发（payload 含 step/error），任务保持 merged、可重调 /pr；跳过时不发 测试：host 29、cloud 469 全过。 --- cloud/README.md | 22 +- cloud/db.py | 5 +- cloud/sandbox.py | 231 ++++++++++++++++++++- cloud/server.py | 79 +++++++- cloud/tests/test_queue_sandbox.py | 168 +++++++++++++++- cloud/tests/test_server_goal_notify.py | 267 +++++++++++++++++++++++++ cloud/tests/test_worker.py | 151 ++++++++++++++ cloud/worker.py | 69 +++++-- 8 files changed, 965 insertions(+), 27 deletions(-) create mode 100644 cloud/tests/test_server_goal_notify.py diff --git a/cloud/README.md b/cloud/README.md index de1f8a9..22dc337 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -227,8 +227,10 @@ webhook 实测）。 webhook 投 JSON（event_type、project_id、task_id、payload）， slack 投 Slack incoming webhook 格式的一行文本（任务 id + 事件 + 项目，失败附 error 摘要）。任务终态（completed / - failed）由 worker 在 run_one 的终态写回后自动通知订阅者； - 手动分发走 `POST /admin/push/dispatch`，同样经内置发送端。 + failed / goal-unmet）由 worker 在 run_one 的终态写回后自动 + 通知订阅者；PR 端点在建 PR 成功时发 review-requested、 + 落盘失败时发 merge-failed；手动分发走 + `POST /admin/push/dispatch`，同样经内置发送端。 显式注入的 sender 优先于内置端。 - webpush：需要 VAPID 密钥对（私钥绝不进仓库），且需要 浏览器端订阅（endpoint + p256dh + auth，现有订阅模型只有 @@ -364,7 +366,7 @@ psycopg，写明“需安装 psycopg”），绝不静默回落到 SQLite， | `GET /health` | 无需认证 | `200 {"ok": true, "service": "cloud-control-plane"}` | | `POST /projects` | 认证即可 | body `{name, id?}`；无 id 则生成 uuid；调用者记为 owner（调 `AuthStore.create_project`，同时写 projects 表并记审计 `create_project`）；返回 `201` + 项目 | | `GET /projects` | 认证即可 | 只返回调用者是成员的项目：`{"projects": [...]}` | -| `POST /tasks` | `create_task` | body `{project_id, prompt, depends_on?, priority?, idempotency_key?}`；无权限 `403`；同 `project_id + idempotency_key` 已存在则返回已有任务（`200`，不建重复）；否则建 `queued` 任务并记审计 `create_task`，返回 `201` | +| `POST /tasks` | `create_task` | body `{project_id, prompt, depends_on?, priority?, idempotency_key?, executor?, goal?, verify_command?, goal_max_rounds?}`；无权限 `403`；同 `project_id + idempotency_key` 已存在则返回已有任务（`200`，不建重复）；否则建 `queued` 任务并记审计 `create_task`，返回 `201`；`verify_command` 非空即 goal 模式（见下文“goal 模式”），`goal_max_rounds` 缺省 3、须 ≥1 | | `GET /tasks?project_id=...` | 该项目 `read` | 返回该项目任务列表：`{"tasks": [...]}`；无权限 `403` | | `GET /tasks/{id}` | 该任务所属项目 `read` | 无权限 `403`；不存在 `404` | | `POST /tasks/{id}/cancel` | 该项目 `create_task` | `queued` / `running` 改为 `interrupted` 并记审计 `cancel_task`；已是终态（completed / failed / interrupted / merged / merge-failed）返回 `400`；不存在 `404` | @@ -386,7 +388,7 @@ psycopg，写明“需安装 psycopg”），绝不静默回落到 SQLite， | `POST /projects/{id}/push/subscriptions` | 该项目 `create_task` | 记订阅 body `{channel, target, preference?}`，channel 为 webhook/slack/webpush/apns/fcm，preference 为 instant/digest/silent，返回 `201` | | `GET /projects/{id}/push/subscriptions` | 该项目 `read` | 只列调用者自己的订阅（不泄露他人 target） | | `DELETE /projects/{id}/push/subscriptions/{sid}` | 该项目 `create_task` | 只能删自己的订阅，项目 owner 可删任意；不存在 `404`，他人订阅 `403` | -| `POST /admin/push/dispatch` | 该项目 `read` | body `{event_type, project_id, payload?}`；只发状态变化事件（completed/failed/goal-unmet/merge-failed/review-requested）；sender=None 时用内置真实发送端（webhook/slack 真 POST，webpush/apns/fcm 报缺凭证），都不静默成功；worker 在任务终态（completed/failed）时自动调分发，不依赖手动调本端点 | +| `POST /admin/push/dispatch` | 该项目 `read` | body `{event_type, project_id, payload?}`；只发状态变化事件（completed/failed/goal-unmet/merge-failed/review-requested）；sender=None 时用内置真实发送端（webhook/slack 真 POST，webpush/apns/fcm 报缺凭证），都不静默成功；自动触发：worker 在任务终态时调分发（completed/failed/goal-unmet，见“goal 模式”），PR 端点在建 PR 成功时发 review-requested、落盘失败时发 merge-failed，不依赖手动调本端点 | | `POST /webhooks/github` | GitHub 签名 | 见“webhook 端点” | | `POST /webhooks/slack` | Slack 签名 | 见“webhook 端点” | @@ -823,6 +825,18 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳 rowcount == 1）；认领日志与审计里含 executor。注意 `_executor_name` 的修正：CloudExecutor 以前会被算成 local， 现在明确返回 cloud，否则 cloud worker 会认领 local 池的任务。 +- goal 模式（`POST /tasks` 带 `verify_command` 即开启，本地 + host.py `goal` 循环的云端版）：worker 不走普通 dispatch， + 改调执行器的 goal_task（local/Docker 均已接，CloudExecutor + 没有 goal 概念、goal 任务派给它会如实失败，不静默降级）； + 循环内每轮 turn 后跑验收命令、失败输出喂回下一轮，最多 + `goal_max_rounds` 轮（缺省 3，与 host.py 默认一致），goal + 文本缺省用 prompt。达标时任务 completed；未达标时任务 + failed（错误文本为最后一轮验收输出），且只发 goal-unmet + 推送、不再重复发 failed。goal 模式下 worker 级重试与模型 + fallback 关闭（循环内部已重试，再叠加只会污染错误文本）； + goal 任务失败后不要用 `/retry` 同任务重跑——host 任务名由 + 任务 id 派生，重跑会撞“任务已存在”而失败，应新建任务。 - CloudExecutor 通用 runner 接口（厂商无关，`cloud/sandbox.py`）： 本轮不假装接通某个云厂商（ECS、Cloud Run 等专属适配明确不在 范围）。CloudExecutor 接受一个 runner endpoint（显式参数或 diff --git a/cloud/db.py b/cloud/db.py index 359ddfc..901f39c 100644 --- a/cloud/db.py +++ b/cloud/db.py @@ -805,7 +805,7 @@ POSTGRES_SCHEMA = ( "created_at DOUBLE PRECISION, " "repo_url TEXT, " "auto_pr INTEGER)", - # tasks 含 server.migrate_tasks_table 补的那 6 列，建表时一次建齐； + # tasks 含 server.migrate_tasks_table 补的列，建表时一次建齐； # 老库迁移仍走 migrate_tasks_table（PRAGMA 翻译后在 PG 上是 no-op）。 "CREATE TABLE IF NOT EXISTS tasks(" "id TEXT PRIMARY KEY, " @@ -828,6 +828,9 @@ POSTGRES_SCHEMA = ( "executor TEXT, " "pr_url TEXT, " "pr_number INTEGER, " + "goal TEXT, " + "verify_command TEXT, " + "goal_max_rounds INTEGER, " "UNIQUE(project_id, idempotency_key))", "CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status)", "CREATE TABLE IF NOT EXISTS summaries(" diff --git a/cloud/sandbox.py b/cloud/sandbox.py index 53bb4b0..d6192ea 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -33,6 +33,59 @@ class Executor: def execute(self, task) -> dict: raise NotImplementedError("子类必须实现 execute(task)") + def goal_task(self, task) -> dict: + """目标循环（host.py goal）：turn→验测→喂回重试。 + + 子类实现；默认不支持（CloudExecutor 的通用 HTTP 协议里 + 没有 goal/验测概念，调它会如实失败，不静默降级成普通 + dispatch）。结果 dict 除 execute() 的字段外另含： + goal_met（True/False/None=无结论）、goal_error。 + """ + raise NotImplementedError("该执行器不支持 goal 模式") + + +def _parse_goal_output(stdout): + """解析 host.py goal 的 stdout，返回 (met, goal_error)。 + + host.py goal 两种结局都 exit 0，结论只看末尾打印的 + "goal met/unmet after rounds"，不能沿用普通结果解析 + （它无 JSON 时一律 completed，会把 unmet 误判成完成）。 + met True=达标，False=未达标，None=输出里无结论。 + goal_error 取最后一轮的 verify_out（无则取 stdout 尾部）。 + """ + if not stdout: + return None, None + if "goal met after rounds" in stdout: + return True, None + if "goal unmet after rounds" in stdout: + last_verify = None + rounds = 0 + for line in stdout.splitlines(): + if line.startswith("round "): + rounds += 1 + if "verify_out=" in line: + last_verify = line.split("verify_out=", 1)[1] + err = "goal verification failed" + if rounds: + err += " after %d rounds" % rounds + detail = (last_verify or "").strip()[:500] + if detail: + err += ": " + detail + return False, err + return None, None + + +def _goal_stdout_tail(stdout): + """goal 结果存库的 stdout：结论在末尾，截尾不截头。 + + 普通结果截头（_truncate_stdout）会把末尾的结论行截掉， + goal 输出必须留尾部，否则 met/unmet 结论在库里查不到。 + """ + text = str(stdout or "") + if len(text) > MAX_STDOUT_CHARS: + return text[-MAX_STDOUT_CHARS:] + return text + def _truncate_stdout(text): if text is None: @@ -184,13 +237,88 @@ class LocalExecutor(Executor): return self._result_from_runner( task["id"], self.runner(cmd)) + def goal_task(self, task) -> dict: + """调 host.py goal：目标循环（turn→验测→喂回重试）。 + + 命令：host.py [--config CFG] [--model M] goal --name + --task --verify <验收命令> + [--max-rounds N] [--repo ]。goal 文本缺省用 prompt； + 轮次缺省 3，与 host.py 默认一致；repo 取任务的 repo 字段， + 没有则取构造时传的 repo。 + 结论解析见 _goal_result_from_runner：host.py goal 两种结局 + 都 exit 0，不能沿用 _result_from_runner。 + """ + if not isinstance(task, dict) or task.get("id") is None: + raise ValueError("任务缺少 id") + verify = task.get("verify_command") + if not verify: + raise ValueError("goal 任务缺少 verify_command") + goal_text = task.get("goal") or task.get("prompt") + if not goal_text: + raise ValueError("任务缺少 prompt") + rounds = task.get("goal_max_rounds") or 3 + cmd = self._base_command(model=task.get("model")) + cmd += ["goal", "--name", self._host_name(task), + "--task", goal_text, "--verify", verify, + "--max-rounds", str(rounds)] + repo = task.get("repo") or self.repo + if repo: + cmd += ["--repo", repo] + return self._goal_result_from_runner( + task["id"], self.runner(cmd)) + + def _goal_result_from_runner(self, task_id, result): + """把 host.py goal 的 runner 返回值解析成 goal 结果 dict。 + + 解析用完整 stdout（结论在末尾），存库的 stdout 才截尾； + returncode 非 0 一律 failed；为 0 时无结论也判 failed， + 绝不把“没打印结论”当成达标。host 任务已存在时（同任务 + /retry 撞名，goal 没有 run 续跑语义）错误写明请新建任务。 + """ + returncode = getattr(result, "returncode", 1) + stdout_full = getattr(result, "stdout", "") or "" + stderr = getattr(result, "stderr", "") or "" + met, goal_error = _parse_goal_output(stdout_full) + combined = stdout_full + stderr + if "已存在" in combined: + met, goal_error = None, ( + "host 任务已存在：goal 任务不支持同任务重试，" + "请新建任务") + status = "failed" + elif returncode != 0: + status = "failed" + if met is None: + goal_error = ((stderr.strip()[-500:] if stderr.strip() + else stdout_full.strip()[-500:]) + or "host goal failed") + elif met is True: + status = "completed" + elif met is False: + status = "failed" + else: + status = "failed" + goal_error = goal_error or "host goal 输出无结论" + return { + "executor": "local", + "task_id": task_id, + "status": status, + "stdout": _goal_stdout_tail(stdout_full), + "stderr": stderr, + "returncode": returncode, + "host_status": ("goal-met" if met is True else + "goal-unmet" if met is False else None), + "goal_met": met, + "goal_error": goal_error, + } + def popen_task(self, task, kind="dispatch"): """进程组启动 host.py 子进程，返回 Popen（供真取消用）。 start_new_session=True 让子进程自成进程组，worker 取消时 killpg 才能连同子孙一起杀掉。注意：只有默认 runner（真实 subprocess）时才用本方法；单测注入的假 runner 走同步路径， - 不走这里。 + 不走这里。kind=goal 时建 goal 命令（结论解析由 worker 的 + _execute_via_process 按 kind 分发到 _goal_result_from_runner）。 """ if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") @@ -200,6 +328,20 @@ class LocalExecutor(Executor): if kind == "run": cmd += ["run", "--task", self._host_name(task), "--prompt", task["prompt"]] + elif kind == "goal": + verify = task.get("verify_command") + if not verify: + raise ValueError("goal 任务缺少 verify_command") + goal_text = task.get("goal") or task.get("prompt") + if not goal_text: + raise ValueError("任务缺少 prompt") + rounds = task.get("goal_max_rounds") or 3 + cmd += ["goal", "--name", self._host_name(task), + "--task", goal_text, "--verify", verify, + "--max-rounds", str(rounds)] + repo = task.get("repo") or self.repo + if repo: + cmd += ["--repo", repo] else: cmd += ["dispatch", "--name", self._host_name(task), "--task", task["prompt"]] @@ -625,6 +767,93 @@ class DockerExecutor(Executor): network=network, extra_env=extra_env) + def goal_task(self, task, timeout=None, extra_mounts=None, + network=None, extra_env=None) -> dict: + """容器内调 host.py goal：目标循环（turn→验测→喂回重试）。 + + 不走 _run_in_container：它经 _parse_host_result 截头 + stdout，而 goal 结论在末尾，截头会把结论截掉；这里直接 + 调 build_run_command + _call_runner，结论用完整 stdout + 解析，存库的 stdout 截尾。超时与异常处理与 + _run_in_container 同款（超时先 kill+rm 容器）。 + """ + if not isinstance(task, dict) or task.get("id") is None: + raise ValueError("任务缺少 id") + verify = task.get("verify_command") + if not verify: + raise ValueError("goal 任务缺少 verify_command") + goal_text = task.get("goal") or task.get("prompt") + if not goal_text: + raise ValueError("任务缺少 prompt") + rounds = task.get("goal_max_rounds") or 3 + inner = self._inner_host_command( + "goal", "--name", self._host_name(task), + "--task", goal_text, "--verify", verify, + "--max-rounds", str(rounds), model=task.get("model")) + repo = task.get("repo") + if repo: + inner += ["--repo", repo] + cmd = self.build_run_command(task, inner, + extra_mounts=extra_mounts, + network=network, + extra_env=extra_env) + try: + result = self._call_runner(cmd, timeout=timeout) + except Exception as exc: + if exc.__class__.__name__ == "TimeoutExpired": + self.kill_container(task) + return { + "executor": "docker", "task_id": task["id"], + "status": "failed", "stdout": "docker run timeout", + "returncode": None, "host_status": None, "stderr": "", + "container": self.container_name(task), + "goal_met": None, "goal_error": "docker run timeout", + } + return { + "executor": "docker", "task_id": task["id"], + "status": "failed", + "stdout": _truncate_stdout(str(exc)), + "returncode": None, "host_status": None, "stderr": "", + "container": self.container_name(task), + "goal_met": None, "goal_error": str(exc)[:500], + } + returncode = getattr(result, "returncode", 1) + stdout_full = getattr(result, "stdout", "") or "" + stderr = getattr(result, "stderr", "") or "" + met, goal_error = _parse_goal_output(stdout_full) + combined = stdout_full + stderr + if "已存在" in combined: + met, goal_error = None, ( + "host 任务已存在：goal 任务不支持同任务重试，" + "请新建任务") + status = "failed" + elif returncode != 0: + status = "failed" + if met is None: + goal_error = ((stderr.strip()[-500:] if stderr.strip() + else stdout_full.strip()[-500:]) + or "host goal failed") + elif met is True: + status = "completed" + elif met is False: + status = "failed" + else: + status = "failed" + goal_error = goal_error or "host goal 输出无结论" + return { + "executor": "docker", + "task_id": task["id"], + "status": status, + "stdout": _goal_stdout_tail(stdout_full), + "stderr": stderr, + "returncode": returncode, + "host_status": ("goal-met" if met is True else + "goal-unmet" if met is False else None), + "container": self.container_name(task), + "goal_met": met, + "goal_error": goal_error, + } + def network_gateway(self, network_name): """某 Docker 网络的网关 IP（宿主在该网桥上的地址）。 diff --git a/cloud/server.py b/cloud/server.py index 29da652..13c9166 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -40,6 +40,7 @@ import argparse import hashlib import html import json +import logging import os import sqlite3 import threading @@ -114,6 +115,10 @@ from cloud.webhooks import ( verify_slack_signature, ) +# PR 事件推送失败时记日志（systemd 下进 journal）；handler 的 +# log_message 被重写为静默（测试输出干净），所以不用它。 +_server_log = logging.getLogger("cloud.server") + TERMINAL_STATUSES = frozenset( {"completed", "failed", "interrupted", "merged", "merge-failed"} ) @@ -245,9 +250,11 @@ def migrate_tasks_table(conn): """幂等迁移：给老库 tasks 表补新列，不丢数据。 新列：result_text、error_text、actual_model、started_at、 - finished_at、child_pid。PRAGMA table_info 查现有列，缺哪个 - 补哪个；已迁移的库再跑一次是 no-op。worker 写执行结果、 - server 读任务时都依赖这些列存在。 + finished_at、child_pid、executor、pr_url、pr_number，以及 + goal、verify_command、goal_max_rounds（goal 模式用）。 + PRAGMA table_info 查现有列，缺哪个补哪个；已迁移的库再跑 + 一次是 no-op。worker 写执行结果、server 读任务时都依赖 + 这些列存在。 """ cols = {row[1] for row in conn.execute( "PRAGMA table_info(tasks)").fetchall()} @@ -261,6 +268,9 @@ def migrate_tasks_table(conn): "executor": "TEXT", "pr_url": "TEXT", "pr_number": "INTEGER", + "goal": "TEXT", + "verify_command": "TEXT", + "goal_max_rounds": "INTEGER", } for name, coltype in new_cols.items(): if name not in cols: @@ -392,6 +402,9 @@ def _task_to_dict(row): "executor": _row_get(row, "executor"), "pr_url": _row_get(row, "pr_url"), "pr_number": _row_get(row, "pr_number"), + "goal": _row_get(row, "goal"), + "verify_command": _row_get(row, "verify_command"), + "goal_max_rounds": _row_get(row, "goal_max_rounds"), } @@ -1222,6 +1235,29 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(400, { "error": "executor must be local, docker or cloud"}) return + # goal 字段：verify_command 非空即 goal 模式（worker 走 + # host.py goal 循环）；goal 文本缺省用 prompt；轮次缺省 3 + # （与 host.py 默认一致），非法值 400 且不建任务。 + goal = body.get("goal") + if goal is not None and not isinstance(goal, str): + self._send_json(400, {"error": "goal must be a string"}) + return + verify_command = body.get("verify_command") + if verify_command is not None and not isinstance( + verify_command, str): + self._send_json(400, { + "error": "verify_command must be a string"}) + return + try: + goal_max_rounds = int(body.get("goal_max_rounds", 3)) + except (TypeError, ValueError): + self._send_json(400, { + "error": "goal_max_rounds must be an integer"}) + return + if goal_max_rounds < 1: + self._send_json(400, { + "error": "goal_max_rounds must be >= 1"}) + return idempotency_key = body.get("idempotency_key") if idempotency_key is not None and not isinstance(idempotency_key, str): self._send_json(400, {"error": "idempotency_key must be a string"}) @@ -1247,11 +1283,13 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): conn.execute( "INSERT INTO tasks(id, project_id, prompt, status," " idempotency_key, depends_on, priority, created_by," - " created_at, updated_at, last_heartbeat, executor)" - " VALUES(?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, NULL, ?)", + " created_at, updated_at, last_heartbeat, executor," + " goal, verify_command, goal_max_rounds)" + " VALUES(?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, NULL, ?," + " ?, ?, ?)", (task_id, project_id, prompt, idempotency_key, json.dumps(depends_on), priority, user_id, now, now, - executor), + executor, goal, verify_command, goal_max_rounds), ) conn.commit() except (sqlite3.IntegrityError, _PgIntegrityError): @@ -1499,6 +1537,25 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(200, worker) # -- 自动 PR（默认关闭，显式触发）---------------------------------- + def _notify_pr_event(self, project_id, event_type, payload): + """PR 事件推送：best-effort，失败只记日志，绝不破坏 PR 响应。 + + review-requested = PR 建成功（PR 在等评审）； + merge-failed = 已 merged 任务的落盘（push/建 PR）失败， + 任务状态保持 merged、可重调 /pr——事件说的是落盘失败， + 不是状态变 merge-failed（云端 merge 端点只改状态， + 没有真合并可失败）。 + """ + try: + result = _push_dispatch( + self.server.db_path, event_type, project_id, payload) + except Exception as exc: # noqa: BLE001 -- 通知失败绝不炸 PR + _server_log.warning("pr notify %s failed: %s", event_type, exc) + return + if result.get("errors"): + _server_log.warning("pr notify %s errors: %s", + event_type, result["errors"]) + def _task_pr(self, user_id, task_id): """给已 merged 的任务建 PR（见 cloud/pr.py 的跳过规则）。 @@ -1507,6 +1564,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): pr_skipped 与 reason，任务状态保持 merged 不变；push/pr 创建失败时返回 502 并写清哪一步失败，任务同样保持 merged。 成功时把 pr_url/pr_number 写回任务并记审计 pr_created。 + 推送：成功时发 review-requested，落盘失败时发 merge-failed + （均为 best-effort，不影响 PR 响应；跳过时不发）。 App-token：项目 repo_url 能解析出 github.com 仓库名且 库里有该仓库的 App 安装记录时，现换 installation token 走 App 路径（响应里 auth_via=app），否则走原来的 gh + @@ -1582,6 +1641,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): }) return except PrStepError as exc: + self._notify_pr_event( + task["project_id"], "merge-failed", + {"task_id": task_id, "step": exc.step, + "error": exc.message}) self._send_json(502, { "error": exc.message, "step": exc.step, "task_id": task_id, "status": task["status"], @@ -1605,6 +1668,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): with _auth_store(self.server.db_path) as store: store.audit(user_id, task["project_id"], "pr_created", result["pr_url"]) + self._notify_pr_event( + task["project_id"], "review-requested", + {"task_id": task_id, "pr_url": result["pr_url"], + "pr_number": result["pr_number"]}) self._send_json(200, dict(task, pr_url=result["pr_url"], pr_number=result["pr_number"], branch=result["branch"], diff --git a/cloud/tests/test_queue_sandbox.py b/cloud/tests/test_queue_sandbox.py index fe6d082..3535164 100644 --- a/cloud/tests/test_queue_sandbox.py +++ b/cloud/tests/test_queue_sandbox.py @@ -18,7 +18,8 @@ sys.path.insert(0, os.path.dirname(os.path.dirname( os.path.abspath(__file__)))) from queue import TaskQueue, budget_exceeded, order_tasks, ready_tasks -from sandbox import CloudExecutor, DockerExecutor, Executor, LocalExecutor +from sandbox import (CloudExecutor, DockerExecutor, Executor, + LocalExecutor, _parse_goal_output) def _task(task_id, depends_on=None, priority=0, status="queued"): @@ -219,5 +220,170 @@ class TestExecutors(unittest.TestCase): self.assertIn("cloud account", str(ctx.exception)) +class TestGoalExecutors(unittest.TestCase): + def test_parse_goal_output(self): + met, err = _parse_goal_output( + "round 1: turn=completed verify_rc=0 verify_out=ok\n" + "goal met after rounds\n") + self.assertTrue(met) + self.assertIsNone(err) + met, err = _parse_goal_output( + "round 1: turn=completed verify_rc=1 verify_out=file missing\n" + "round 2: turn=completed verify_rc=1 verify_out=file missing\n" + "goal unmet after rounds\n") + self.assertFalse(met) + self.assertIn("file missing", err) + self.assertIn("2 rounds", err) + met, err = _parse_goal_output("some other output\n") + self.assertIsNone(met) + met, err = _parse_goal_output("") + self.assertIsNone(met) + + def test_local_goal_task_met(self): + seen = [] + + def fake(cmd): + seen.append(cmd) + return SimpleNamespace( + returncode=0, + stdout="round 1: turn=completed verify_rc=0 verify_out=ok\n" + "goal met after rounds\n", + stderr="") + + executor = LocalExecutor("/fake/host.py", runner=fake) + result = executor.goal_task( + {"id": "t1", "prompt": "do it", "verify_command": "test -f x", + "goal_max_rounds": 2, "repo": "/r"}) + self.assertEqual(result["status"], "completed") + self.assertTrue(result["goal_met"]) + self.assertIsNone(result["goal_error"]) + self.assertEqual(result["host_status"], "goal-met") + self.assertIn("goal", seen[0]) + self.assertIn("--verify", seen[0]) + self.assertIn("test -f x", seen[0]) + self.assertIn("/r", seen[0]) + self.assertEqual(seen[0][seen[0].index("--max-rounds") + 1], "2") + + def test_local_goal_task_unmet(self): + def fake(cmd): + return SimpleNamespace( + returncode=0, + stdout="round 1: turn=completed verify_rc=1 " + "verify_out=file missing\n" + "goal unmet after rounds\n", + stderr="") + + result = LocalExecutor("/fake/host.py", + runner=fake).goal_task( + {"id": "t1", "prompt": "do it", + "verify_command": "test -f x"}) + self.assertEqual(result["status"], "failed") + self.assertFalse(result["goal_met"]) + self.assertIn("file missing", result["goal_error"]) + self.assertEqual(result["host_status"], "goal-unmet") + + def test_local_goal_task_defaults_prompt_and_rounds(self): + seen = [] + + def fake(cmd): + seen.append(list(cmd)) + return SimpleNamespace( + returncode=0, stdout="goal met after rounds\n", stderr="") + + LocalExecutor("/fake/host.py", runner=fake).goal_task( + {"id": "t1", "prompt": "do it", "verify_command": "true"}) + self.assertIn("do it", seen[0]) + self.assertEqual(seen[0][seen[0].index("--max-rounds") + 1], "3") + + def test_local_goal_task_no_conclusion_is_failed_not_completed(self): + def fake(cmd): + return SimpleNamespace(returncode=0, stdout="nothing\n", + stderr="") + + result = LocalExecutor("/fake/host.py", + runner=fake).goal_task( + {"id": "t1", "prompt": "do it", + "verify_command": "true"}) + self.assertEqual(result["status"], "failed") + self.assertIsNone(result["goal_met"]) + self.assertIn("无结论", result["goal_error"]) + + def test_local_goal_task_nonzero_is_failed(self): + def fake(cmd): + return SimpleNamespace(returncode=1, stdout="", stderr="boom") + + result = LocalExecutor("/fake/host.py", + runner=fake).goal_task( + {"id": "t1", "prompt": "do it", + "verify_command": "true"}) + self.assertEqual(result["status"], "failed") + self.assertIn("boom", result["goal_error"]) + + def test_local_goal_task_already_exists_names_retry(self): + def fake(cmd): + return SimpleNamespace( + returncode=1, stdout="", + stderr="Traceback\nRuntimeError: 任务已存在") + + result = LocalExecutor("/fake/host.py", + runner=fake).goal_task( + {"id": "t1", "prompt": "do it", + "verify_command": "true"}) + self.assertEqual(result["status"], "failed") + self.assertIn("请新建任务", result["goal_error"]) + + def test_local_goal_task_requires_verify(self): + with self.assertRaises(ValueError): + LocalExecutor("/fake/host.py", + runner=lambda cmd: None).goal_task( + {"id": "t1", "prompt": "do it"}) + + def test_base_goal_task_not_implemented(self): + with self.assertRaises(NotImplementedError): + Executor().goal_task({"id": "t1"}) + + def test_popen_goal_command(self): + seen = [] + + class _FakePopen: + def __init__(self, cmd, **kwargs): + seen.append(list(cmd)) + + with mock.patch("sandbox.subprocess.Popen", _FakePopen): + LocalExecutor("/fake/host.py").popen_task( + {"id": "t1", "prompt": "do it", + "verify_command": "test -f x", "goal_max_rounds": 2}, + kind="goal") + self.assertIn("goal", seen[0]) + self.assertIn("test -f x", seen[0]) + self.assertEqual(seen[0][seen[0].index("--max-rounds") + 1], "2") + + def test_docker_goal_task_unmet(self): + seen = [] + + def fake_runner(cmd, timeout=None): + seen.append(cmd) + return SimpleNamespace( + returncode=0, + stdout="round 1: turn=completed verify_rc=1 " + "verify_out=file missing\n" + "goal unmet after rounds\n", + stderr="") + + with mock.patch("sandbox.shutil.which", + return_value="/usr/bin/docker"): + executor = DockerExecutor( + "/fake/host.py", runner=fake_runner, + env={"XCM_API_KEY": "placeholder-not-a-real-key"}) + result = executor.goal_task( + {"id": "t1", "prompt": "do it", + "verify_command": "test -f x"}) + self.assertEqual(result["executor"], "docker") + self.assertEqual(result["status"], "failed") + self.assertFalse(result["goal_met"]) + self.assertIn("goal", seen[0]) + self.assertIn("test -f x", seen[0]) + + if __name__ == "__main__": unittest.main() diff --git a/cloud/tests/test_server_goal_notify.py b/cloud/tests/test_server_goal_notify.py new file mode 100644 index 0000000..082b501 --- /dev/null +++ b/cloud/tests/test_server_goal_notify.py @@ -0,0 +1,267 @@ +"""goal 字段与 PR 事件推送的端点级单测。 + +- POST /tasks 接受 goal/verify_command/goal_max_rounds，非法值 400。 +- init_db 迁移后 tasks 表含 goal 三列。 +- POST /tasks/{id}/pr 成功时发 review-requested，落盘失败时发 + merge-failed（订阅指向本地回环监听，断言真收到）。 +git/gh 一律 fake，不跑真实命令、不联网。 +""" + +import http.client +import http.server +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.push import create_subscription + from cloud.server import create_server, init_db +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.push import create_subscription + from cloud.server import create_server, init_db + + +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, push_rc=0): + self.calls = [] + self.remote = remote + self.branch = branch + self.push_rc = push_rc + + 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(self.push_rc, "", + "push rejected" if self.push_rc else "") + 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 _Listener: + """本地回环 webhook 接收器。""" + + def __init__(self): + self.requests = [] + + outer = self + + class Handler(http.server.BaseHTTPRequestHandler): + def do_POST(self): + length = int(self.headers.get("Content-Length") or 0) + body = self.rfile.read(length) if length else b"" + outer.requests.append(json.loads(body.decode("utf-8"))) + self.send_response(200) + self.send_header("Content-Length", "2") + self.end_headers() + self.wfile.write(b"{}") + + def log_message(self, *args): + pass + + self.server = http.server.ThreadingHTTPServer( + ("127.0.0.1", 0), Handler) + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + self.url = "http://127.0.0.1:%d/hook" % \ + self.server.server_address[1] + + def stop(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + + +class GoalPrNotifyTestCase(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.listener = _Listener() + + def tearDown(self): + self.listener.stop() + 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="p"): + status, project = self._req( + "POST", "/projects", {"name": name, "auto_pr": 1}, + self.owner_key) + self.assertEqual(status, 201, project) + return project["id"] + + def _merged_task(self, project_id, task_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_create_task_goal_fields(self): + pid = self._project() + status, task = self._req( + "POST", "/tasks", + {"project_id": pid, "prompt": "do it", + "goal": "make x", "verify_command": "test -f x", + "goal_max_rounds": 2}, + self.owner_key) + self.assertEqual(status, 201, task) + self.assertEqual(task["goal"], "make x") + self.assertEqual(task["verify_command"], "test -f x") + self.assertEqual(task["goal_max_rounds"], 2) + # 不带 goal 字段：普通任务，字段为空。 + status, task = self._req( + "POST", "/tasks", + {"project_id": pid, "prompt": "plain"}, + self.owner_key) + self.assertEqual(status, 201, task) + self.assertIsNone(task["goal"]) + self.assertIsNone(task["verify_command"]) + # 非法轮次：0 与非数字都 400，且不建任务。 + for bad in (0, -1, "x"): + status, body = self._req( + "POST", "/tasks", + {"project_id": pid, "prompt": "bad", + "verify_command": "true", "goal_max_rounds": bad}, + self.owner_key) + self.assertEqual(status, 400, body) + + def test_migration_adds_goal_columns(self): + conn = sqlite3.connect(self.db_path) + try: + cols = {row[1] for row in conn.execute( + "PRAGMA table_info(tasks)").fetchall()} + finally: + conn.close() + self.assertIn("goal", cols) + self.assertIn("verify_command", cols) + self.assertIn("goal_max_rounds", cols) + + def test_pr_success_notifies_review_requested(self): + pid = self._project() + create_subscription( + self.db_path, pid, self.owner_id, "webhook", + self.listener.url, "instant") + self._merged_task(pid, "t-pr-ok") + self.server.pr_git_runner = _Git( + "https://github.com/octo/repo.git", "host/t-pr-ok") + self.server.pr_gh_runner = _Gh( + "https://github.com/octo/repo/pull/9") + status, body = self._req( + "POST", "/tasks/t-pr-ok/pr", {}, self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(len(self.listener.requests), 1) + event = self.listener.requests[0] + self.assertEqual(event["event_type"], "review-requested") + self.assertEqual(event["task_id"], "t-pr-ok") + self.assertIn("pull/9", event["payload"]["pr_url"]) + + def test_pr_failure_notifies_merge_failed(self): + pid = self._project() + create_subscription( + self.db_path, pid, self.owner_id, "webhook", + self.listener.url, "instant") + self._merged_task(pid, "t-pr-fail") + self.server.pr_git_runner = _Git( + "https://github.com/octo/repo.git", "host/t-pr-fail", + push_rc=1) + self.server.pr_gh_runner = _Gh( + "https://github.com/octo/repo/pull/9") + status, body = self._req( + "POST", "/tasks/t-pr-fail/pr", {}, self.owner_key) + self.assertEqual(status, 502, body) + self.assertEqual(body["step"], "push") + self.assertEqual(len(self.listener.requests), 1) + event = self.listener.requests[0] + self.assertEqual(event["event_type"], "merge-failed") + self.assertEqual(event["task_id"], "t-pr-fail") + self.assertEqual(event["payload"]["step"], "push") + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_worker.py b/cloud/tests/test_worker.py index e1f9075..990ca70 100644 --- a/cloud/tests/test_worker.py +++ b/cloud/tests/test_worker.py @@ -223,6 +223,157 @@ class WorkerTestCase(unittest.TestCase): server.server_close() thread.join(timeout=5) + # -- goal 模式 ------------------------------------------------------- + def _set_goal(self, task_id, verify="test -f x", goal=None, + rounds=None): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "UPDATE tasks SET verify_command = ?, goal = ?," + " goal_max_rounds = ? WHERE id = ?", + (verify, goal, rounds, task_id)) + conn.commit() + finally: + conn.close() + + def test_goal_met_notifies_completed(self): + """goal 达标：任务 completed，发 completed（不发 goal-unmet）。""" + import http.server + import threading + + requests = [] + + class Handler(http.server.BaseHTTPRequestHandler): + def do_POST(self): + length = int(self.headers.get("Content-Length") or 0) + body = self.rfile.read(length) if length else b"" + requests.append(json.loads(body.decode("utf-8"))) + self.send_response(200) + self.send_header("Content-Length", "2") + self.end_headers() + self.wfile.write(b"{}") + + def log_message(self, *args): + pass + + server = http.server.ThreadingHTTPServer( + ("127.0.0.1", 0), Handler) + thread = threading.Thread( + target=server.serve_forever, daemon=True) + thread.start() + try: + from cloud.push import create_subscription + url = "http://127.0.0.1:%d/hook" % server.server_address[1] + create_subscription( + self.db_path, "p1", "user-1", "webhook", url, "instant") + _insert_task(self.db_path, "A") + self._set_goal("A") + calls = [] + + def goal_runner(cmd): + calls.append(list(cmd)) + return SimpleNamespace( + returncode=0, + stdout="round 1: turn=completed verify_rc=0 " + "verify_out=ok\ngoal met after rounds\n", + stderr="") + + worker = self._worker(runner=goal_runner) + results = worker.run_all() + self.assertEqual( + [r["status"] for r in results], ["completed"]) + self.assertIn("goal", calls[0]) + self.assertEqual(len(requests), 1) + self.assertEqual(requests[0]["event_type"], "completed") + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + def test_goal_unmet_notifies_only_goal_unmet(self): + """goal 未达：任务 failed，只发 goal-unmet，不重复发 failed。""" + import http.server + import threading + + requests = [] + + class Handler(http.server.BaseHTTPRequestHandler): + def do_POST(self): + length = int(self.headers.get("Content-Length") or 0) + body = self.rfile.read(length) if length else b"" + requests.append(json.loads(body.decode("utf-8"))) + self.send_response(200) + self.send_header("Content-Length", "2") + self.end_headers() + self.wfile.write(b"{}") + + def log_message(self, *args): + pass + + server = http.server.ThreadingHTTPServer( + ("127.0.0.1", 0), Handler) + thread = threading.Thread( + target=server.serve_forever, daemon=True) + thread.start() + try: + from cloud.push import create_subscription + url = "http://127.0.0.1:%d/hook" % server.server_address[1] + create_subscription( + self.db_path, "p1", "user-1", "webhook", url, "instant") + _insert_task(self.db_path, "A") + self._set_goal("A") + + def unmet_runner(cmd): + return SimpleNamespace( + returncode=0, + stdout="round 1: turn=completed verify_rc=1 " + "verify_out=file missing\n" + "goal unmet after rounds\n", + stderr="") + + worker = self._worker(runner=unmet_runner) + results = worker.run_all() + self.assertEqual([r["status"] for r in results], ["failed"]) + self.assertEqual(len(requests), 1) + self.assertEqual(requests[0]["event_type"], "goal-unmet") + self.assertIn("file missing", + requests[0]["payload"]["error"]) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + def test_goal_mode_runs_once_despite_retries(self): + """goal 模式只跑一次：循环内部已多轮重试，worker 级重试 + 不再叠加（同任务重跑会撞 host 任务名已存在）。""" + _insert_task(self.db_path, "A") + self._set_goal("A") + calls = [] + + def counting(cmd): + calls.append(cmd) + return SimpleNamespace( + returncode=0, + stdout="round 1: turn=completed verify_rc=1 " + "verify_out=nope\ngoal unmet after rounds\n", + stderr="") + + worker = self._worker(runner=counting, max_retries=2) + results = worker.run_all() + self.assertEqual(len(calls), 1) + self.assertEqual(results[0]["status"], "failed") + + def test_cloud_executor_goal_unsupported(self): + """cloud 执行器没有 goal 概念：goal 任务如实失败， + 不静默降级成普通 dispatch。""" + from cloud.sandbox import CloudExecutor + _insert_task(self.db_path, "A") + self._set_goal("A") + worker = self._worker(executor=CloudExecutor()) + results = worker.run_all() + self.assertEqual(results[0]["status"], "failed") + self.assertIn("不支持 goal", results[0]["error"]) + # -- 取消 ------------------------------------------------------------ def test_cancel_queued_task_marks_interrupted(self): _insert_task(self.db_path, "A") diff --git a/cloud/worker.py b/cloud/worker.py index 87774ba..0c8e6ad 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -214,6 +214,12 @@ def _task_from_row(row): # 老库行缺 executor 列时给 None（表示任意 worker 可认领）， # 不炸；列本身由 _ensure_result_columns 在写回时补。 "executor": _col("executor"), + # goal 字段同款 _col 容错：老库行与 load_tasks 的显式列清单 + # 里没有时给 None（= 普通任务），不炸；run_one 认领后 + # get_task 重读 SELECT *，goal 任务一定能拿到完整字段。 + "goal": _col("goal"), + "verify_command": _col("verify_command"), + "goal_max_rounds": _col("goal_max_rounds"), } @@ -422,7 +428,8 @@ class Worker: rows = conn.execute( "SELECT id, project_id, prompt, status, idempotency_key," " depends_on, priority, created_by, created_at, updated_at," - " last_heartbeat, executor FROM tasks" + " last_heartbeat, executor, goal, verify_command," + " goal_max_rounds FROM tasks" " WHERE status IN ('queued', 'completed', 'merged')" " ORDER BY created_at, id").fetchall() finally: @@ -511,7 +518,9 @@ class Worker: "actual_model": "TEXT", "started_at": "REAL", "finished_at": "REAL", "child_pid": "INTEGER", "executor": "TEXT", "pr_url": "TEXT", - "pr_number": "INTEGER"}.items(): + "pr_number": "INTEGER", "goal": "TEXT", + "verify_command": "TEXT", + "goal_max_rounds": "INTEGER"}.items(): if name not in cols: try: conn.execute( @@ -644,8 +653,9 @@ class Worker: error=None): """任务终态推送：best-effort，绝不因通知失败影响任务状态。 - 只发 completed/failed（push.ALLOWED_PUSH_EVENTS 的子集； - interrupted 不在事件集合里，取消不通知）。无订阅时 + 只发终态事件：completed/failed/goal-unmet + （push.ALLOWED_PUSH_EVENTS 的子集；interrupted 不在 + 事件集合里，取消不通知）。无订阅时 dispatch 直接返回空 sent，不联网。同步发送、单次 10 秒 超时：慢订阅会拖住 worker 循环最多几秒，换取投递结果 可观测（失败记日志，不静默丢）；后台队列是后续优化， @@ -921,6 +931,9 @@ class Worker: "host_status": None} fake = SimpleNamespace(returncode=proc.returncode, stdout=out or "", stderr=err or "") + if kind == "goal" and hasattr( + self.executor, "_goal_result_from_runner"): + return self.executor._goal_result_from_runner(task_id, fake) if hasattr(self.executor, "_result_from_runner"): return self.executor._result_from_runner(task_id, fake) return {"executor": "local", "task_id": task_id, @@ -1114,6 +1127,10 @@ class Worker: task_id = task["id"] result = None error = None + # goal 模式：任务带 verify_command 时走 host.py goal 循环， + # 不走普通 dispatch。goal 循环内部已多轮重试，worker 级 + # 的已存在续跑（run）对它没有意义（goal 没有 run 语义）。 + goal_mode = bool(task.get("verify_command")) # Docker 同步路径：dispatch 阻塞在 docker run 里时，取消 # 线程靠容器名杀容器（见 cancel()）。容器名按任务派生， # 与执行器本身无关，所以这里只记名不碰执行器状态。 @@ -1142,10 +1159,11 @@ class Worker: try: if self._use_popen(): result = self._execute_via_process( - task, deadline_seconds=self.task_timeout) + task, kind="goal" if goal_mode else "dispatch", + deadline_seconds=self.task_timeout) combined = (result.get("stdout", "") or "") + \ (result.get("stderr", "") or "") - if result["status"] not in ( + if not goal_mode and result["status"] not in ( "completed", "cancelled", "timeout") and \ _ALREADY_EXISTS_MARKER in combined: result = self._execute_via_process( @@ -1209,16 +1227,26 @@ class Worker: "NO_PROXY": "localhost,127.0.0.1", "no_proxy": "localhost,127.0.0.1", } - result = self.executor.dispatch_task( - task, timeout=effective_docker_timeout, - extra_mounts=tenant_mounts, - network=tenant_network, - extra_env=proxy_env) + if goal_mode: + result = self.executor.goal_task( + task, timeout=effective_docker_timeout, + extra_mounts=tenant_mounts, + network=tenant_network, + extra_env=proxy_env) + else: + result = self.executor.dispatch_task( + task, timeout=effective_docker_timeout, + extra_mounts=tenant_mounts, + network=tenant_network, + extra_env=proxy_env) else: - result = self.executor.dispatch_task(task) + if goal_mode: + result = self.executor.goal_task(task) + else: + result = self.executor.dispatch_task(task) combined = (result.get("stdout", "") or "") + \ (result.get("stderr", "") or "") - if result["status"] != "completed" and \ + if not goal_mode and result["status"] != "completed" and \ _ALREADY_EXISTS_MARKER in combined: if is_docker: result = self.executor.run_task( @@ -1432,6 +1460,11 @@ class Worker: if self._was_cancelled(task_id) or \ result.get("status") == "cancelled": break + if task.get("verify_command"): + # goal 模式只跑一次：循环内部已多轮重试（turn→验测→ + # 喂回），worker 级重试/fallback 不再叠加——同任务重跑 + # 会撞 host 任务名已存在，第二次尝试必失败且污染错误文本。 + break if result.get("status") == "completed": break host_status_try = self._fallback_token(result, error) @@ -1517,6 +1550,8 @@ class Worker: % (self.model, first_error, fallback_model, second_error)) else: + if result.get("goal_error"): + error = result["goal_error"] if error is None: stderr = (result or {}).get("stderr", "") or "" error = stdout.strip() or \ @@ -1569,7 +1604,13 @@ class Worker: else "worker_failed", task_id=task_id, project_id=task["project_id"], status=final) - if final in ("completed", "failed"): + if result.get("goal_met") is False: + # goal 未达只发 goal-unmet，不再重复发 failed： + # 同一结局两条通知会刷屏，digest 偏好下两者又都是 + # 即时事件，更不能双发。 + self._notify_task_finished( + task_id, task["project_id"], "goal-unmet", error=error) + elif final in ("completed", "failed"): self._notify_task_finished( task_id, task["project_id"], final, error=error) else: