feat: 旧任务继续会话——终端任务带新消息 + 全部选项同一会话续跑

server:POST /tasks/{id}/continue(终端非 merged 任务重排队,
resume_message 列存本轮消息,不传选项沿用原值,plan/goal 冲突
与新建一致,goal_mode=1 时转新 goal 语义;CAS 防并发改状态,
审计 task_continued);tasks.resume_message 三处迁移 + PG 建表。
worker:认领后本轮 prompt 换成续跑消息(只换内存副本),
goal 续跑走 goal-resume 而不是 goal,跑完清 resume_message
(进程被杀时不清,下次认领重跑同一条消息)。
host:goal-resume --goal-text(旧任务接续 goal 模式:跳过状态
与记录守卫,旧线程上从第 1 轮开新目标循环,新语义不限轮次;
无线程时拒绝)。
sandbox:Local/Docker goal_resume_task + popen goal-resume kind;
附带 popen goal 命令改成新语义(无验收不传 --verify,轮次 0
原样传,此前 goal 任务到本地 worker 会直接报错)。
console:详情页终端非 merged 任务出继续会话表单(消息 +
模型/强度/权限/计划/goal,模型下拉复用中转列表)。
测试:新增 test_task_continue.py 11 个 + test_host_continue.py 2 个。
This commit is contained in:
host committed 2026-10-27 04:40:11 +08:00
1 parent cf293914d9
commit 3a4997d346
9 files changed
+823 -20

No files matched your search

+18
View File
@@ -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
+56 -4
View File
@@ -215,6 +215,30 @@
<input id="forkMsg" type="text" placeholder="分支首轮消息(必填,带历史的新会话)" style="width:100%; box-sizing:border-box; margin:6px 0;">
<div class="bar"><button id="btnForkConfirm" class="primary">创建分支任务</button><span id="forkMsgSpan" class="muted"></span></div>
</div>
<div id="continueForm" style="display:none">
<h3>继续会话(旧任务接回来,同一会话再跑一轮;不传的选项沿用任务原值)</h3>
<textarea id="continueText" placeholder="本轮要说的话(必填)"></textarea>
<div class="bar">
<select id="continueModel"><option value="">模型:沿用任务原值</option></select>
<select id="continueEffort">
<option value="">强度:沿用任务原值</option>
<option value="low">low</option>
<option value="medium">medium</option>
<option value="high">high</option>
<option value="xhigh">xhigh</option>
<option value="max">max</option>
</select>
<select id="continueApproval">
<option value="">权限:沿用任务原值</option>
<option value="ask">请求批准</option>
<option value="auto">帮我批准</option>
<option value="full">完全访问</option>
</select>
<label><input type="checkbox" id="continuePlan"> 计划模式(只读)</label>
<label><input type="checkbox" id="continueGoal"> goal 模式(不限轮次,模型宣告完成)</label>
</div>
<div class="bar"><button id="btnContinueConfirm" class="primary">继续会话</button><span id="continueMsgSpan" class="muted"></span></div>
</div>
<pre id="detailBody" style="display:none"></pre>
<h3>实时链路</h3>
<div id="traceBox" class="muted">点任务行查看详情后,这里显示过程链路</div>
@@ -280,10 +304,12 @@ async function loadRelayModels() {
try {
var d = await api("/relay/models", "GET");
var models = d.models || [];
sel.innerHTML = "<option value=''>模型:默认(跟随配置)</option>" +
models.map(function (m) {
return "<option value='" + esc(m) + "'>" + esc(m) + "</option>";
}).join("");
var opts = models.map(function (m) {
return "<option value='" + esc(m) + "'>" + esc(m) + "</option>";
}).join("");
sel.innerHTML = "<option value=''>模型:默认(跟随配置)</option>" + opts;
var csel = $("#continueModel");
if (csel) csel.innerHTML = "<option value=''>模型:沿用任务原值</option>" + 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 = "<span class='err'>加载失败:" + esc(e.message) + "</span>"; }
@@ -723,6 +751,30 @@ $("#btnForkConfirm").onclick = async function () {
loadTasks();
} catch (e) { $("#forkMsgSpan").innerHTML = "<span class='err'>分支失败:" + esc(e.message) + "</span>"; }
};
$("#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 = "<span class='ok'>已重新排队,同一会话继续</span>";
$("#continueText").value = "";
showDetail(currentTask.id);
loadTasks();
} catch (e) { $("#continueMsgSpan").innerHTML = "<span class='err'>继续失败:" + esc(e.message) + "</span>"; }
};
$("#btnQueueFollowup").onclick = function () { postFollowup("queued"); };
$("#btnDirectFollowup").onclick = function () { postFollowup("direct"); };
$("#btnCost").onclick = async function () {
+3 -1
View File
@@ -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))
+110 -6
View File
@@ -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 <host 名>
--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(宿主在该网桥上的地址)。
+158
View File
@@ -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):
"""把终态失败任务重新排队:只改状态,不直接执行。
+315
View File
@@ -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()
+50 -9
View File
@@ -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,
+27
View File
@@ -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} 当前状态是 "
+86
View File
@@ -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()