diff --git a/cloud/README.md b/cloud/README.md index c48567b..5f3f96e 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -495,6 +495,24 @@ verify_command，goal 新语义（无验收）任务到本地 worker 会直接 报错，本批顺手修成与同步 goal_task 一致（无验收不传 --verify， 轮次 0 原样传）。 +## 旧任务继续会话（终端任务接回来再跑一轮） + +详情页对终端任务（completed/failed/interrupted/merge-failed） +提供“继续会话”：填本轮消息 + 模型/强度/权限/计划/goal 全可选 +（不传的沿用任务原值），`POST /tasks/{id}/continue` 把任务重 +新排队，worker 认领后在同一会话线程上再跑一轮——不是新任务 +（原 prompt 不动，只换本轮输入），也不是分支（不建新线程）。 +新消息存在 tasks.resume_message 列，worker 跑完清掉；进程中途 +被杀时清不到，下次认领重跑同一条消息，不静默丢消息。 +goal 模式的继续：server 把任务转成新 goal 语义（goal=新消息、 +验收清空、轮次 0），worker 走 host.py `goal-resume --goal-text` +在旧线程上开新目标循环（不限轮次，模型宣告完成才结束）； +普通任务的继续走 dispatch 撞已存在 → run 续跑（与重试同款）。 +不适用：进行中任务请用后续消息（queued/running 各 409），已 +合并任务 worktree 已删、请分支（merged 409）；冲突与新建一致 +（plan + approval=full、plan + goal 各 400）。 + + ## 测试 ```bash diff --git a/cloud/console.html b/cloud/console.html index f8ca93f..99db94c 100644 --- a/cloud/console.html +++ b/cloud/console.html @@ -215,6 +215,30 @@
+

实时链路

点任务行查看详情后，这里显示过程链路
@@ -280,10 +304,12 @@ async function loadRelayModels() { try { var d = await api("/relay/models", "GET"); var models = d.models || []; - sel.innerHTML = "" + - models.map(function (m) { - return ""; - }).join(""); + var opts = models.map(function (m) { + return ""; + }).join(""); + sel.innerHTML = "" + opts; + var csel = $("#continueModel"); + if (csel) csel.innerHTML = "" + opts; $("#modelCustom").style.display = "none"; } catch (e) { $("#modelCustom").style.display = "inline"; @@ -529,6 +555,8 @@ async function showDetail(taskId) { $("#forkMsgSpan").textContent = ""; $("#followupMsg").textContent = ""; $("#followupForm").style.display = ["queued", "running"].indexOf(t.status) >= 0 ? "block" : "none"; + $("#continueMsgSpan").textContent = ""; + $("#continueForm").style.display = (TRACE_TERMINAL.indexOf(t.status) >= 0 && t.status !== "merged") ? "block" : "none"; startTrace(taskId, t.status); loadFollowups(taskId); } catch (e) { $("#detail").innerHTML = "加载失败：" + esc(e.message) + ""; } @@ -723,6 +751,30 @@ $("#btnForkConfirm").onclick = async function () { loadTasks(); } catch (e) { $("#forkMsgSpan").innerHTML = "分支失败：" + esc(e.message) + ""; } }; +$("#btnContinueConfirm").onclick = async function () { + if (!currentTask) return; + var msg = $("#continueText").value.trim(); + if (!msg) { $("#continueMsgSpan").textContent = "本轮消息必填"; return; } + var body = { message: msg }; + var model = $("#continueModel").value; + if (model) body.model = model; + var eff = $("#continueEffort").value; + if (eff) body.effort = eff; + var ap = $("#continueApproval").value; + if (ap) { + if (ap === "full" && !confirm("完全访问 = 自动批准 + 不受工作区限制，agent 可读写工作区外路径。确认继续吗？")) return; + body.approval = ap; + } + if ($("#continuePlan").checked) body.plan = 1; + if ($("#continueGoal").checked) body.goal_mode = 1; + try { + var t = await api("/tasks/" + encodeURIComponent(currentTask.id) + "/continue", "POST", body); + $("#continueMsgSpan").innerHTML = "已重新排队，同一会话继续"; + $("#continueText").value = ""; + showDetail(currentTask.id); + loadTasks(); + } catch (e) { $("#continueMsgSpan").innerHTML = "继续失败：" + esc(e.message) + ""; } +}; $("#btnQueueFollowup").onclick = function () { postFollowup("queued"); }; $("#btnDirectFollowup").onclick = function () { postFollowup("direct"); }; $("#btnCost").onclick = async function () { diff --git a/cloud/db.py b/cloud/db.py index ae18cfd..c7afcbe 100644 --- a/cloud/db.py +++ b/cloud/db.py @@ -838,6 +838,7 @@ POSTGRES_SCHEMA = ( "plan INTEGER, " "forked_from TEXT, " "environment_id TEXT, " + "resume_message TEXT, " "UNIQUE(project_id, idempotency_key))", "CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status)", # 附件包与任务附件：与 server._ensure_uploads_tables 同款， @@ -965,7 +966,8 @@ def init_postgres_schema(conn): ("pr_number", "INTEGER"), ("worktree_host", "TEXT"), ("goal_mode", "INTEGER"), - ("environment_id", "TEXT")): + ("environment_id", "TEXT"), + ("resume_message", "TEXT")): if _name not in tcols: conn.execute( "ALTER TABLE tasks ADD COLUMN %s %s" % (_name, _type)) diff --git a/cloud/sandbox.py b/cloud/sandbox.py index e5bdb69..8de55f3 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -386,6 +386,27 @@ class LocalExecutor(Executor): return self._goal_result_from_runner( task["id"], self.runner(cmd)) + def goal_resume_task(self, task, goal_text) -> dict: + """调 host.py goal-resume --goal-text：在旧线程上开新目标 + 循环（旧任务接续 goal 模式，云端 /continue 带 goal_mode=1 + 时走这里）。命令：host.py goal-resume --task + --goal-text <新目标> --max-rounds 0（新语义不限轮次）。 + 不传 --repo：goal-resume 操作已有 host 任务，不新建 + worktree。结论解析与 goal_task 同款。 + """ + if not isinstance(task, dict) or task.get("id") is None: + raise ValueError("任务缺少 id") + if not goal_text: + raise ValueError("goal-resume 缺少目标文本") + cmd = self._base_command(model=task.get("model"), + effort=task.get("effort"), + approval=task.get("approval")) + cmd += ["goal-resume", "--task", self._host_name(task), + "--goal-text", goal_text, + "--max-rounds", "0"] + 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。 @@ -437,12 +458,14 @@ class LocalExecutor(Executor): start_new_session=True 让子进程自成进程组，worker 取消时 killpg 才能连同子孙一起杀掉。注意：只有默认 runner（真实 subprocess）时才用本方法；单测注入的假 runner 走同步路径， - 不走这里。kind=goal 时建 goal 命令（结论解析由 worker 的 - _execute_via_process 按 kind 分发到 _goal_result_from_runner）。 - goal 语义与同步 goal_task 一致：新形态（goal_mode）无验收 - 命令时 --verify 不传、--max-rounds 传 0（0 不能用 or 兜底 - 成 3）；旧形态仍传验收命令与轮次。seed_dir 只对 dispatch/ - goal 追加（run 续跑不传，文件已在 worktree 里）。 + 不走这里。kind=goal 时建 goal 命令、kind=goal-resume 时建 + goal-resume 命令（旧任务接续 goal 模式，目标文本取本轮 + prompt；结论解析由 worker 的 _execute_via_process 按 kind + 分发到 _goal_result_from_runner）。goal 语义与同步 goal_task + 一致：新形态（goal_mode）无验收命令时 --verify 不传、 + --max-rounds 传 0（0 不能用 or 兜底成 3）；旧形态仍传验收 + 命令与轮次。seed_dir 只对 dispatch/goal 追加（run 续跑与 + goal-resume 不传，文件已在 worktree 里）。 """ if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") @@ -474,6 +497,13 @@ class LocalExecutor(Executor): cmd += ["--repo", repo] if seed_dir: cmd += ["--seed-dir", seed_dir] + elif kind == "goal-resume": + # 旧任务接续 goal：host 侧用 --goal-text 在旧线程上开 + # 新目标循环（新语义，不限轮次）；task["prompt"] 在续跑 + # 时已被 worker 换成续跑消息。 + cmd += ["goal-resume", "--task", self._host_name(task), + "--goal-text", task["prompt"], + "--max-rounds", "0"] else: cmd += ["dispatch", "--name", self._host_name(task), "--task", task["prompt"]] @@ -1041,6 +1071,80 @@ class DockerExecutor(Executor): "usage": _parse_goal_usage(stdout_full), } + def goal_resume_task(self, task, goal_text, timeout=None, + extra_mounts=None, network=None, + extra_env=None) -> dict: + """容器内调 host.py goal-resume --goal-text：在旧线程上开 + 新目标循环（旧任务接续 goal 模式）。解析与 goal_task 同款 + （完整 stdout 结论、存库截尾），但没有“已存在”分支： + goal-resume 本来就是操作已有 host 任务，不存在 dispatch + 撞名问题；returncode 非 0 或无结论一律 failed。 + """ + if not isinstance(task, dict) or task.get("id") is None: + raise ValueError("任务缺少 id") + if not goal_text: + raise ValueError("goal-resume 缺少目标文本") + inner = self._inner_host_command( + "goal-resume", "--task", self._host_name(task), + "--goal-text", goal_text, "--max-rounds", "0", + model=task.get("model"), effort=task.get("effort"), + approval=task.get("approval")) + 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) + if returncode != 0: + status = "failed" + if met is None: + goal_error = ((stderr.strip()[-500:] if stderr.strip() + else stdout_full.strip()[-500:]) + or "host goal-resume failed") + elif met is True: + status = "completed" + elif met is False: + status = "failed" + else: + status = "failed" + goal_error = goal_error or "host goal-resume 输出无结论" + 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, + "usage": _parse_goal_usage(stdout_full), + } + def network_gateway(self, network_name): """某 Docker 网络的网关 IP（宿主在该网桥上的地址）。 diff --git a/cloud/server.py b/cloud/server.py index ec5101f..21a1a99 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -466,6 +466,10 @@ def migrate_tasks_table(conn): # 项目仓库路径）；环境本身是项目仓库的一个命名基准检出 # （分支 + 执行器），任务 worktree 从它建，不改任务隔离。 "environment_id": "TEXT", + # 续跑消息：旧任务继续会话时本轮要说的话（NULL = 没有待 + # 续跑消息）。worker 认领后把 prompt 换成它跑一轮，跑完 + # 清掉；任务详情里照常显示原 prompt，续跑消息只作本轮输入。 + "resume_message": "TEXT", } for name, coltype in new_cols.items(): if name not in cols: @@ -777,6 +781,7 @@ def _task_to_dict(row, attachments=None): "goal_max_rounds": _row_get(row, "goal_max_rounds"), "goal_mode": _row_get(row, "goal_mode") or 0, "attachments": list(attachments or []), + "resume_message": _row_get(row, "resume_message"), "worktree_host": _row_get(row, "worktree_host"), "model": _row_get(row, "model"), "effort": _row_get(row, "effort"), @@ -1132,6 +1137,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._list_followups(user_id, rest[: -len("/followups")]) elif rest.endswith("/fork") and method == "POST": self._fork_task(user_id, rest[: -len("/fork")]) + elif rest.endswith("/continue") and method == "POST": + self._continue_task(user_id, rest[: -len("/continue")]) else: self._send_json(404, {"error": "not found"}) else: @@ -2615,6 +2622,157 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): store.audit(user_id, task["project_id"], "merge_task", task_id) self._send_json(200, task) + def _continue_task(self, user_id, task_id): + """旧任务接回来继续会话：终端任务（completed/failed/ + interrupted/merge-failed）带一条新消息 + 全部选项重排队， + worker 认领后在同一会话线程上再跑一轮（不是新任务、不 + 是分支）。新消息存在 resume_message 列，worker 跑完清掉； + 原 prompt 不动，详情里仍显示原 prompt。 + + 规则：任务必须存在（404）、调用方须有 create_task 权限 + （与后续消息同级：这本质上是“给旧会话发消息”）、任务 + 必须是终端且非 merged（queued/running 请用后续消息， + merged 的 worktree 已删、请分支，各 409）。选项与新建 + 任务同款校验：model（空串 = 跟随配置）、effort、approval、 + plan、goal_mode；不传的选项沿用任务原值。冲突与新建 + 一致：plan + approval=full、plan + goal_mode 各 400。 + goal_mode=1 时同时把任务转成新 goal 语义（goal=新消息、 + verify_command 清空、轮次 0），worker 走 goal-resume + 在旧线程上开新目标循环；goal_mode 不传时 goal 任务沿用 + 原语义续跑。状态回 queued 时上一轮 result/error 清空 + （与重试一致，否则排队中的任务还挂着旧结果）。 + 并发下用 CAS（WHERE status IN 终端四态），没改到行 + 说明状态已变，返回 409 而不是假装成功。写审计 + task_continued。 + """ + row = self._fetch_task(task_id) + if row is None: + self._send_json(404, {"error": "task not found"}) + return + task = _task_to_dict(row) + with _auth_store(self.server.db_path) as store: + if not store.check_permission( + user_id, task["project_id"], "create_task"): + self._send_json(403, {"error": "forbidden"}) + return + if task["status"] in ("queued", "running"): + self._send_json(409, { + "error": "进行中任务请用后续消息继续", + "status": task["status"], + }) + return + if task["status"] == "merged": + self._send_json(409, { + "error": "已合并任务不能继续（worktree 已删），请分支", + "status": task["status"], + }) + return + if task["status"] not in TERMINAL_STATUSES: + self._send_json(409, { + "error": "only terminal tasks can be continued", + "status": task["status"], + }) + return + body, err = self._read_json() + if err: + self._send_json(400, {"error": err}) + return + message = body.get("message") + if not isinstance(message, str) or not message.strip(): + self._send_json(400, {"error": "message 必填"}) + return + message = message.strip() + if len(message) > 20000: + self._send_json(400, {"error": "message too long"}) + return + 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 + 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 + 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 模式冲突"}) + return + sets = ["status = 'queued'", "resume_message = ?", + "result_text = NULL", "error_text = NULL", + "actual_model = NULL", "updated_at = ?"] + vals = [message, time.time()] + if model is not None or "model" in body: + sets.append("model = ?") + vals.append(model) + if effort is not None: + sets.append("effort = ?") + vals.append(effort) + if approval is not None: + sets.append("approval = ?") + vals.append(approval) + if plan is not None: + sets.append("plan = ?") + vals.append(plan) + if goal_mode is not None: + sets.append("goal_mode = ?") + vals.append(goal_mode) + if goal_mode == 1: + sets.append("goal = ?") + vals.append(message) + sets.append("verify_command = NULL") + sets.append("goal_max_rounds = 0") + vals.append(task_id) + conn = self._db() + try: + cur = conn.execute( + "UPDATE tasks SET %s WHERE id = ? AND status IN" + " ('completed', 'failed', 'interrupted', 'merge-failed')" + % ", ".join(sets), + tuple(vals), + ) + conn.commit() + if cur.rowcount != 1: + self._send_json(409, { + "error": "任务状态已变化，请刷新后重试", + "status": task["status"], + }) + return + new_row = conn.execute( + "SELECT * FROM tasks WHERE id = ?", (task_id,) + ).fetchone() + finally: + conn.close() + with _auth_store(self.server.db_path) as store: + store.audit(user_id, task["project_id"], "task_continued", + task_id) + self._send_json(200, _task_to_dict(new_row)) + def _retry_task(self, user_id, task_id): """把终态失败任务重新排队：只改状态，不直接执行。 diff --git a/cloud/tests/test_task_continue.py b/cloud/tests/test_task_continue.py new file mode 100644 index 0000000..8b479b2 --- /dev/null +++ b/cloud/tests/test_task_continue.py @@ -0,0 +1,315 @@ +"""旧任务继续会话单测：终端任务带新消息 + 全部选项重排队， +worker 在同一会话线程上再跑一轮（不是新任务、不是分支）。 + +不跑真实模型回合、不联网：server 侧走本地临时端口 + 临时 +SQLite；worker/sandbox 侧注入假 runner 录命令。 +""" + +import http.client +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import time +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + from cloud.auth import AuthStore + from cloud.sandbox import DockerExecutor, LocalExecutor + from cloud.server import create_server, init_db + from cloud.worker import Worker +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.sandbox import DockerExecutor, LocalExecutor + from cloud.server import create_server, init_db + from cloud.worker import Worker + + +class ServerContinueTestCase(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.other_id, self.other_key = self.auth.create_user("other") + 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() + status, self.project = self._req( + "POST", "/projects", {"name": "demo"}, self.owner_key) + self.assertEqual(status, 201) + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + + def _req(self, method, path, body=None, key=None): + conn = http.client.HTTPConnection( + "127.0.0.1", self.port, timeout=10) + hdrs = {} + payload = None + if key: + hdrs["X-API-Key"] = key + if body is not None: + payload = json.dumps(body) + hdrs["Content-Type"] = "application/json" + conn.request(method, path, body=payload, headers=hdrs) + resp = conn.getresponse() + raw = resp.read() + conn.close() + try: + return resp.status, json.loads(raw.decode("utf-8")) + except (ValueError, UnicodeDecodeError): + return resp.status, {} + + def _task_in_status(self, status, **extra): + _s, task = self._req( + "POST", "/tasks", + {"project_id": self.project["id"], "prompt": "original"}, + self.owner_key) + self.assertEqual(_s, 201, task) + conn = sqlite3.connect(self.db_path) + sets = ["status = ?"] + vals = [status] + for k, v in extra.items(): + sets.append("%s = ?" % k) + vals.append(v) + vals.append(task["id"]) + conn.execute("UPDATE tasks SET %s WHERE id = ?" + % ", ".join(sets), vals) + conn.commit() + conn.close() + return task["id"] + + def _continue(self, task_id, body, key=None): + return self._req( + "POST", "/tasks/%s/continue" % task_id, body, + key or self.owner_key) + + def test_continue_completed_with_options(self): + tid = self._task_in_status("completed") + status, task = self._continue(tid, { + "message": "follow up please", "model": "gpt-6-sol", + "effort": "high", "approval": "auto", "plan": 1}) + self.assertEqual(status, 200, task) + self.assertEqual(task["status"], "queued") + self.assertEqual(task["resume_message"], "follow up please") + self.assertEqual(task["model"], "gpt-6-sol") + self.assertEqual(task["effort"], "high") + self.assertEqual(task["approval"], "auto") + self.assertEqual(task["plan"], 1) + self.assertEqual(task["prompt"], "original") + + def test_continue_keeps_options_when_absent(self): + tid = self._task_in_status("failed", model="gpt-5.6-terra") + status, task = self._continue(tid, {"message": "again"}) + self.assertEqual(status, 200, task) + self.assertEqual(task["model"], "gpt-5.6-terra") + self.assertEqual(task["resume_message"], "again") + + def test_continue_goal_mode_converts_to_new_semantics(self): + tid = self._task_in_status( + "completed", goal="old goal", verify_command="true", + goal_max_rounds=3) + status, task = self._continue( + tid, {"message": "new goal", "goal_mode": 1}) + self.assertEqual(status, 200, task) + self.assertEqual(task["goal_mode"], 1) + self.assertEqual(task["goal"], "new goal") + self.assertIsNone(task["verify_command"]) + self.assertEqual(task["goal_max_rounds"], 0) + + def test_continue_rejects_non_terminal_and_merged(self): + for status in ("queued", "running"): + tid = self._task_in_status(status) + code, _body = self._continue(tid, {"message": "hi"}) + self.assertEqual(code, 409, status) + tid = self._task_in_status("merged") + code, _body = self._continue(tid, {"message": "hi"}) + self.assertEqual(code, 409) + code, _body = self._continue("no-such-task", {"message": "hi"}) + self.assertEqual(code, 404) + + def test_continue_rejects_bad_input_and_conflicts(self): + tid = self._task_in_status("completed") + for body, code in [ + ({"message": ""}, 400), + ({"message": "x" * 20001}, 400), + ({"message": "hi", "effort": "ultra"}, 400), + ({"message": "hi", "approval": "never"}, 400), + ({"message": "hi", "plan": 1, "approval": "full"}, 400), + ({"message": "hi", "plan": 1, "goal_mode": 1}, 400)]: + got, _body = self._continue(tid, body) + self.assertEqual(got, code, body) + + def test_continue_forbidden_for_non_member(self): + tid = self._task_in_status("completed") + code, _body = self._continue(tid, {"message": "hi"}, + key=self.other_key) + self.assertEqual(code, 403) + + +class WorkerContinueTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + self.env = {"XCM_API_KEY": "test-placeholder"} + + def tearDown(self): + self.tmp.cleanup() + + def _insert(self, task_id, **extra): + now = time.time() + cols = ["id", "project_id", "prompt", "status", + "idempotency_key", "depends_on", "priority", + "created_by", "created_at", "updated_at", + "last_heartbeat"] + vals = [task_id, "p1", "original", "queued", None, "[]", 0, + "tester", now, now, None] + for k, v in extra.items(): + cols.append(k) + vals.append(v) + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO tasks(%s) VALUES(%s)" + % (", ".join(cols), ", ".join("?" * len(cols))), vals) + conn.commit() + finally: + conn.close() + + def _row(self, task_id): + conn = sqlite3.connect(self.db_path) + try: + row = conn.execute( + "SELECT status, resume_message FROM tasks WHERE id = ?", + (task_id,)).fetchone() + finally: + conn.close() + return row + + def test_resume_normal_runs_new_message_on_same_thread(self): + self._insert("t1", resume_message="follow up please") + seen = [] + + def fake(cmd): + seen.append(list(cmd)) + if "dispatch" in cmd: + # host 任务已存在：dispatch 失败，worker 回落到 run。 + return SimpleNamespace(returncode=1, stdout="任务已存在", + stderr="") + if "run" in cmd: + return SimpleNamespace(returncode=0, stdout="ok", + stderr="") + raise AssertionError("不应有第三种命令: %s" % cmd) + + worker = Worker(self.db_path, "/fake/host.py", runner=fake, + env=self.env) + result = worker.run_one() + self.assertEqual(result["status"], "completed") + run_cmds = [c for c in seen if "run" in c] + self.assertTrue(run_cmds) + self.assertEqual(run_cmds[0][run_cmds[0].index("--prompt") + 1], + "follow up please") + status, resume = self._row("t1") + self.assertEqual(status, "completed") + self.assertIsNone(resume) + + def test_resume_goal_routes_to_goal_resume(self): + self._insert("t2", goal_mode=1, goal="old goal", + goal_max_rounds=0, + resume_message="new goal") + seen = [] + + def fake(cmd): + seen.append(list(cmd)) + if "goal-resume" in cmd: + return SimpleNamespace( + returncode=0, + stdout="round 1: turn=completed verify_rc=- " + "verify_out=\ngoal met after rounds\n", + stderr="") + raise AssertionError("goal 续跑不应走 goal/dispatch: %s" % cmd) + + worker = Worker(self.db_path, "/fake/host.py", runner=fake, + env=self.env) + result = worker.run_one() + self.assertEqual(result["status"], "completed") + self.assertTrue(any("goal-resume" in c for c in seen)) + cmd = [c for c in seen if "goal-resume" in c][0] + self.assertEqual(cmd[cmd.index("--goal-text") + 1], "new goal") + self.assertEqual(cmd[cmd.index("--max-rounds") + 1], "0") + status, resume = self._row("t2") + self.assertEqual(status, "completed") + self.assertIsNone(resume) + + +class SandboxContinueTestCase(unittest.TestCase): + def _recording(self, calls): + def fake(cmd): + calls.append(list(cmd)) + return SimpleNamespace(returncode=0, stdout="ok", stderr="") + return fake + + def test_local_goal_resume_command(self): + calls = [] + ex = LocalExecutor("/fake/host.py", + runner=self._recording(calls)) + ex.goal_resume_task({"id": "t1", "prompt": "new goal"}, + "new goal") + self.assertIn("goal-resume", calls[0]) + self.assertEqual(calls[0][calls[0].index("--goal-text") + 1], + "new goal") + self.assertEqual(calls[0][calls[0].index("--max-rounds") + 1], + "0") + + def test_docker_goal_resume_command(self): + from unittest import mock + calls = [] + ex = DockerExecutor( + "/fake/host.py", runner=self._recording(calls), + env={"XCM_API_KEY": "placeholder-not-a-real-key"}) + with mock.patch("cloud.sandbox.shutil.which", + return_value="/usr/bin/docker"): + ex.goal_resume_task({"id": "t1", "prompt": "new goal"}, + "new goal") + self.assertIn("goal-resume", calls[0]) + self.assertIn("new goal", calls[0]) + + def test_popen_goal_resume_command(self): + import cloud.sandbox as sandbox_mod + calls = [] + + class FakePopen(object): + def __init__(self, cmd, **kwargs): + calls.append(list(cmd)) + + orig = sandbox_mod.subprocess.Popen + sandbox_mod.subprocess.Popen = FakePopen + try: + ex = LocalExecutor("/fake/host.py") + ex.popen_task({"id": "t1", "prompt": "new goal"}, + kind="goal-resume") + finally: + sandbox_mod.subprocess.Popen = orig + self.assertIn("goal-resume", calls[0]) + self.assertEqual(calls[0][calls[0].index("--goal-text") + 1], + "new goal") + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 5530b73..733f598 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -247,6 +247,8 @@ def _task_from_row(row): "forked_from": _col("forked_from"), # 环境：老行缺列时 None = 不指定环境，走项目仓库路径。 "environment_id": _col("environment_id"), + # 续跑消息：老行缺列时 None = 没有待续跑消息。 + "resume_message": _col("resume_message"), } @@ -580,7 +582,8 @@ class Worker: "approval": "TEXT", "plan": "INTEGER", "forked_from": "TEXT", - "environment_id": "TEXT"}.items(): + "environment_id": "TEXT", + "resume_message": "TEXT"}.items(): if name not in cols: try: conn.execute( @@ -1018,7 +1021,7 @@ class Worker: "host_status": None} fake = SimpleNamespace(returncode=proc.returncode, stdout=out or "", stderr=err or "") - if kind == "goal" and hasattr( + if kind in ("goal", "goal-resume") 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"): @@ -1634,8 +1637,14 @@ class Worker: effective_docker_timeout = self.task_timeout try: if self._use_popen(): + if goal_mode and task.get("_resume"): + popen_kind = "goal-resume" + elif goal_mode: + popen_kind = "goal" + else: + popen_kind = "dispatch" result = self._execute_via_process( - task, kind="goal" if goal_mode else "dispatch", + task, kind=popen_kind, deadline_seconds=self.task_timeout, seed_host_dir=seed_host_dir) combined = (result.get("stdout", "") or "") + \ @@ -1663,12 +1672,20 @@ class Worker: "stderr": ""}, prep_err) if goal_mode: - result = self.executor.goal_task( - task, timeout=effective_docker_timeout, - extra_mounts=tenant_mounts, - network=tenant_network, - extra_env=proxy_env, - seed_host_dir=seed_host_dir) + if task.get("_resume"): + result = self.executor.goal_resume_task( + task, task["prompt"], + timeout=effective_docker_timeout, + extra_mounts=tenant_mounts, + network=tenant_network, + extra_env=proxy_env) + else: + result = self.executor.goal_task( + task, timeout=effective_docker_timeout, + extra_mounts=tenant_mounts, + network=tenant_network, + extra_env=proxy_env, + seed_host_dir=seed_host_dir) else: result = self.executor.dispatch_task( task, timeout=effective_docker_timeout, @@ -1836,6 +1853,15 @@ class Worker: return {"dispatched": False, "reason": "already_claimed", "task_id": task_id} task = self.get_task(task_id) or task + # 续跑消息：旧任务继续会话时本轮要说的话——本轮 prompt 换 + # 成它（只换内存里的副本，原 prompt 不动）；_resume 标记 + # 让 _run_attempt 在 goal 模式下走 goal-resume 而不是 goal + # （goal 会撞已存在，只有 goal-resume 能在旧线程上开新循环）。 + resume_message = task.get("resume_message") + if resume_message: + task = dict(task) + task["prompt"] = resume_message + task["_resume"] = True self._audit(task["project_id"], "worker_claim", "%s executor=%s" % (task_id, self._executor_name())) # log_event 只接受 task_id/project_id/status/message 四个 @@ -2153,6 +2179,21 @@ class Worker: # 补一条 worker_completed/failed 审计误导排查。 final = "interrupted" self._audit(task["project_id"], "worker_interrupted", task_id) + if resume_message: + # 本轮已跑完（无论成败）：续跑消息已消费，清掉；进程 + # 中途被杀时清不到，下次认领会重跑同一条消息——可接受 + # 的续跑语义，不会静默丢消息。清失败不挡结果返回。 + try: + conn = _db_connect(self.db_path) + try: + conn.execute( + "UPDATE tasks SET resume_message = NULL" + " WHERE id = ?", (task_id,)) + conn.commit() + finally: + conn.close() + except Exception: + pass return {"dispatched": True, "task_id": task_id, "host_name": self.executor._host_name(task) if hasattr(self.executor, "_host_name") else task_id, diff --git a/host.py b/host.py index fd4df36..98ece6f 100644 --- a/host.py +++ b/host.py @@ -1992,6 +1992,10 @@ def main(): help="验测命令上限（秒）；0 表示不限时，默认 600") p = sub.add_parser("goal-resume") p.add_argument("--task", required=True) + p.add_argument("--goal-text", default=None, + help="新目标文本：旧任务接续 goal 模式时用——跳过状态与" + "goal 记录守卫，在旧线程上从第 1 轮开新目标循环" + "（新语义：无验收、不传 --max-rounds 时不限轮次）") p.add_argument("--max-rounds", type=int, default=None, help="不传时沿用原任务的 goal_max_rounds；传了则覆盖总轮次") p.add_argument("--verify-timeout", type=int, default=600, @@ -2583,6 +2587,29 @@ def main(): verify_timeout=args.verify_timeout) elif args.cmd == "goal-resume": t = state["tasks"][args.task] + if args.goal_text: + # 旧任务接续 goal 模式（云端 /continue 带 goal_mode=1 + # 时走这里）：旧任务已终局、没有 goal 记录是正常情况， + # 不能按断点续跑的守卫拒绝。只要线程还在，就在旧线程 + # 上从第 1 轮开新目标循环（新语义：无验收、不传轮次 + # 时不限轮次，模型宣告完成才结束）。 + if not t.get("thread_id"): + print(f"goal-resume 拒绝：{args.task} 没有会话线程，" + f"不能接续") + return + max_rounds = args.max_rounds \ + if args.max_rounds is not None else 0 + srv.resume_thread(t["thread_id"]) + run_goal_loop(srv, state, args.state, t, args.goal_text, + None, max_rounds, + args.model, args.effort, args.sandbox, cfg=cfg, + timeout=args.turn_timeout, + verify_timeout=args.verify_timeout, + start_round=1, + last_verify="", + last_turn_status=None, + last_turn_text="") + return if t.get("status") not in ("goal-running", "interrupted", "running"): print(f"goal-resume 拒绝：{args.task} 当前状态是 " diff --git a/tests/test_host_continue.py b/tests/test_host_continue.py new file mode 100644 index 0000000..b13a241 --- /dev/null +++ b/tests/test_host_continue.py @@ -0,0 +1,86 @@ +"""旧任务接续 goal 模式：host.py goal-resume --goal-text。 + +已终局的旧任务没有 goal 记录是正常情况，--goal-text 模式跳过 +状态与记录守卫，在旧线程上从第 1 轮开新目标循环（新语义： +无验收、不传轮次时不限轮次，模型宣告完成才结束）。 +""" + +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT)) + +import host + + +class FakeServer(object): + def __init__(self, text): + self.text = text + self.resumed = [] + + def initialize(self): + pass + + def close(self): + pass + + def resume_thread(self, thread_id): + self.resumed.append(thread_id) + + def run_turn(self, thread_id, prompt, model=None, effort=None, + sandbox=None, timeout=None): + return {"status": "completed", "text": self.text, + "turn_id": "u1"} + + +class GoalResumeTextTest(unittest.TestCase): + def _run_main(self, task, goal_text): + with tempfile.TemporaryDirectory() as tmp: + state_path = os.path.join(tmp, "host-state.json") + cfg_path = os.path.join(tmp, "host-config.json") + with open(state_path, "w", encoding="utf-8") as f: + json.dump({"tasks": {"g": task}, "automations": []}, f) + with open(cfg_path, "w", encoding="utf-8") as f: + json.dump({}, f) + fake = FakeServer("done\n目标已达成") + argv = ["host.py", "--state", state_path, + "--config", cfg_path, + "goal-resume", "--task", "g"] + if goal_text is not None: + argv += ["--goal-text", goal_text] + with mock.patch.object(sys, "argv", argv): + with mock.patch.object(host, "Server", + lambda **kw: fake): + host.main() + with open(state_path, encoding="utf-8") as f: + saved = json.load(f)["tasks"]["g"] + return saved, fake + + def test_goal_text_init_on_completed_task(self): + task = {"name": "g", "thread_id": "x", "cwd": "/tmp", + "sandbox": None, "max_retries": 0, "retry_count": 0, + "turns": [], "status": "completed"} + saved, fake = self._run_main(task, "new goal") + self.assertEqual(saved["status"], "goal-met") + self.assertEqual(saved["goal"], "new goal") + self.assertEqual(saved["goal_rounds_done"], 1) + self.assertEqual(fake.resumed, ["x"]) + + def test_goal_text_refuses_task_without_thread(self): + task = {"name": "g", "cwd": "/tmp", + "sandbox": None, "max_retries": 0, "retry_count": 0, + "turns": [], "status": "completed"} + saved, fake = self._run_main(task, "new goal") + self.assertEqual(saved["status"], "completed") + self.assertEqual(fake.resumed, []) + + +if __name__ == "__main__": + unittest.main()