From 9e56c1d26686e1f4af1e8bad9e97e774c7eac23c Mon Sep 17 00:00:00 2001 From: host Date: Mon, 2 Nov 2026 17:55:18 +0000 Subject: [PATCH] code-to10 goal live update: POST /tasks/{id}/goal updates a running goal loop at the next round boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Server writes goal + monotonic goal_version + task_goal_history in one transaction (CAS on version, approve_merge only, new-semantic goal tasks only, terminal/merged 409, same-goal noop). Worker mirror thread (30s) and breakpoint-resume claim hook write the update as a signal file next to host-state.json; host run_goal_loop applies it at round boundaries (top of round and before the declaration check, so a declaration made for the old goal in a turn during which the goal changed is voided), with an explicit old->new notice, same thread/worktree, rounds/usage/ budget/failed counters untouched. handler method accounting is a contract update: 239 total / 223 mixins / 16 in server.py. (cherry picked from commit 6841f287219d2b36cfbb8803642344b84dae4377) Reconcile-port: handler_lifecycle.py contents redirected — _update_goal_task method + _goal_update_reject/_goal_signal_status pure functions + goal_live_update import into cloud/server.py (target keeps 6-mixin split); _task_to_dict goal fields into cloud/handler_support.py; migrate columns into cloud/server.py migrate_tasks_table; test patch target handler_lifecycle→server; test_handler_split_conformance.py kept at target version, to be re-pinned for the unified line in a final commit. --- cloud/goal_live_update.py | 195 ++++++++++ cloud/handler_support.py | 5 + cloud/server.py | 121 ++++++- cloud/tests/test_goal_live_update.py | 509 +++++++++++++++++++++++++++ cloud/worker.py | 109 ++++++ host.py | 173 ++++++++- 6 files changed, 1110 insertions(+), 2 deletions(-) create mode 100644 cloud/goal_live_update.py create mode 100644 cloud/tests/test_goal_live_update.py diff --git a/cloud/goal_live_update.py b/cloud/goal_live_update.py new file mode 100644 index 0000000..32c6ba7 --- /dev/null +++ b/cloud/goal_live_update.py @@ -0,0 +1,195 @@ +"""运行中改 goal 的纯函数与信号文件格式。 + +信号文件格式（与 host.py 内联读侧同款契约，改一处必须同步 +另一处，跨模块测试钉住它）： + /goal-updates/.json + {"version": int, "goal": str, "task_id": str | None} +其中 safe_host_name 与 host.py 的 _interrupt_request_path 同款： +只留 [A-Za-z0-9_.-]，其余字符换成 "-"。 + +本模块不 import cloud.server / cloud.worker，只用标准库， +host.py 不直接 import 它（host.py 是可独立部署的脚本，只用 +标准库），host.py 里有一份同格式的内联读侧。 + +语义： +- version 单调递增，服务端每次成功改 goal 就 +1；循环只应用 + version > 已应用版本的信号，过期版本直接丢弃。 +- 信号文件只由 worker 侧写（镜像线程每 30 秒把库里更新的 + goal_version 写成信号；断点续跑认领时钩子先写一次）， + host 循环在轮次边界读、应用后由检查点落盘 + 清理。 +- 信号本身不含签名：它只在 host-state 目录（worker 与 host + 循环共享的同一目录）里流转，调用方是受信的 worker 进程； + 不要把信号文件路径暴露成公开接口。 +""" + +import json +import os +import re +import uuid + +GOAL_UPDATE_DIRNAME = "goal-updates" + + +def safe_host_name(host_name): + """与 host.py _interrupt_request_path 同款的安全化。""" + return re.sub(r"[^A-Za-z0-9_.-]", "-", str(host_name)) + + +def signal_path(state_path, host_name): + """信号文件绝对路径：state 文件同目录下的 goal-updates 子目录。""" + return os.path.join( + os.path.dirname(os.path.abspath(state_path)), + GOAL_UPDATE_DIRNAME, + safe_host_name(host_name) + ".json", + ) + + +def write_goal_update_signal(state_path, host_name, version, goal, + task_id=None): + """原子写信号文件（写临时文件再 os.replace）。 + + version 必须为正整数，goal 必须为非空字符串，否则 + ValueError；调用方（worker 镜像/认领钩子）负责吞异常， + 信号写失败绝不能炸掉任务执行本身。 + """ + if not isinstance(version, int) or isinstance(version, bool) \ + or version < 1: + raise ValueError("goal update version must be a positive int") + if not isinstance(goal, str) or not goal.strip(): + raise ValueError("goal update goal must be a non-empty string") + path = signal_path(state_path, host_name) + os.makedirs(os.path.dirname(path), exist_ok=True) + tmp = "%s.tmp.%s" % (path, os.getpid()) + payload = {"version": version, "goal": goal, "task_id": task_id} + with open(tmp, "w", encoding="utf-8") as fh: + json.dump(payload, fh, ensure_ascii=False) + os.replace(tmp, path) + return path + + +def read_goal_update_signal(state_path, host_name, applied_version): + """读信号，返回 (version, goal) 或 None。 + + 以下情况一律返回 None，绝不抛：文件不存在、JSON 坏、 + version 不是正整数、goal 不是非空字符串、version <= + 已应用版本（过期信号）。调用方按 None 处理即可。 + """ + try: + applied = int(applied_version or 0) + except (TypeError, ValueError): + applied = 0 + try: + with open(signal_path(state_path, host_name), + encoding="utf-8") as fh: + data = json.load(fh) + except Exception: + return None + if not isinstance(data, dict): + return None + version = data.get("version") + goal = data.get("goal") + if not isinstance(version, int) or isinstance(version, bool) \ + or version < 1: + return None + if not isinstance(goal, str) or not goal.strip(): + return None + if version <= applied: + return None + return (version, goal) + + +def consume_goal_update_signal(state_path, host_name, applied_version): + """已应用版本的信号文件尽力删掉（删失败静默）。 + + 只删 version <= 已应用版本的信号；更新的信号（写于应用 + 之后、尚未应用）绝不能被误删。返回是否删掉了文件。 + """ + try: + applied = int(applied_version or 0) + except (TypeError, ValueError): + applied = 0 + path = signal_path(state_path, host_name) + try: + with open(path, encoding="utf-8") as fh: + data = json.load(fh) + except Exception: + return False + if not isinstance(data, dict): + return False + version = data.get("version") + if not isinstance(version, int) or isinstance(version, bool): + return False + if version > applied: + return False + try: + os.remove(path) + except Exception: + return False + return True + + +def build_goal_update_notice(old_goal, new_goal): + """轮次边界应用新 goal 时，喂给 agent 的显式通知块。""" + return ( + "目标已更新（运行中改 goal）：\n" + "旧目标：%s\n" + "新目标：%s\n" + "请继续完成新目标，不要重复已完成的部分；完成与否只按" + "新目标判断；此前如已输出过“目标已达成”，该宣告对新" + "目标作废——新目标满足时请重新宣告，否则继续工作。" + % (old_goal, new_goal) + ) + + +def ensure_goal_history_table(conn): + """幂等建 task_goal_history 表（改 goal 的审计历史）。 + + id 为随机 hex；(task_id, version) 上的索引供按任务查 + 历史时用。调用方负责 commit。 + """ + conn.execute( + "CREATE TABLE IF NOT EXISTS task_goal_history(" + "id TEXT PRIMARY KEY, task_id TEXT NOT NULL, " + "old_goal TEXT, new_goal TEXT NOT NULL, " + "version INTEGER NOT NULL, created_by TEXT, " + "created_at REAL NOT NULL)" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_task_goal_history_task " + "ON task_goal_history(task_id, version)" + ) + + +def apply_goal_update(conn, task_id, old_goal, old_version, new_goal, + user_id, now): + """在同一连接里做一次改 goal 写库（不 commit，调用方提交）。 + + CAS 在 version 上：只有 version 仍是 old_version（老行 + NULL 按 0 计）时才更新，否则返回 None（调用方回 409 请 + 刷新后重试，绝不静默覆盖并发改动）。成功时同时写 + goal_history 一行，返回新 version。 + """ + old_v = int(old_version or 0) + if old_v == 0: + cur = conn.execute( + "UPDATE tasks SET goal = ?, goal_version = ?, " + "updated_at = ? WHERE id = ? AND " + "(goal_version IS NULL OR goal_version = 0)", + (new_goal, old_v + 1, now, task_id), + ) + else: + cur = conn.execute( + "UPDATE tasks SET goal = ?, goal_version = ?, " + "updated_at = ? WHERE id = ? AND goal_version = ?", + (new_goal, old_v + 1, now, task_id, old_v), + ) + if cur.rowcount != 1: + return None + conn.execute( + "INSERT INTO task_goal_history(id, task_id, old_goal, " + "new_goal, version, created_by, created_at) " + "VALUES(?, ?, ?, ?, ?, ?, ?)", + (uuid.uuid4().hex, task_id, old_goal, new_goal, + old_v + 1, user_id, now), + ) + return old_v + 1 diff --git a/cloud/handler_support.py b/cloud/handler_support.py index a479461..ea4d6a7 100644 --- a/cloud/handler_support.py +++ b/cloud/handler_support.py @@ -154,6 +154,11 @@ def _task_to_dict(row, attachments=None, total_tokens=None): "verify_command": _row_get(row, "verify_command"), "goal_max_rounds": _row_get(row, "goal_max_rounds"), "goal_mode": _row_get(row, "goal_mode") or 0, + # 运行中改 goal：goal_version 是已提交版本（服务端写库）， + # applied_goal_version 是 host 循环已应用版本（镜像回写）， + # 两者不等时说明信号在途或未到轮次边界，绝不能冒充已应用。 + "goal_version": _row_get(row, "goal_version") or 0, + "applied_goal_version": _row_get(row, "applied_goal_version") or 0, "attachments": list(attachments or []), "resume_message": _row_get(row, "resume_message"), "worktree_host": _row_get(row, "worktree_host"), diff --git a/cloud/server.py b/cloud/server.py index a5beb18..3c2a1b1 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -74,6 +74,10 @@ from cloud.db import init_postgres_schema as _init_postgres_schema from cloud.db import PostgresIntegrityError as _PgIntegrityError from cloud.db import PostgresOperationalError as _PgOperationalError from cloud.goal_command import parse_goal_command as _parse_goal_command +from cloud.goal_live_update import ( + apply_goal_update as _glu_apply_goal_update, + ensure_goal_history_table as _glu_ensure_history, +) from cloud.peer_messages import deliver_peer_message as _deliver_peer_message from cloud.peer_messages import ensure_followups_table as _peer_ensure_followups from cloud.peer_messages import peer_message_prefix as _peer_prefix @@ -1118,7 +1122,8 @@ def migrate_tasks_table(conn): finished_at、child_pid、executor、pr_url、pr_number，以及 goal、verify_command、goal_max_rounds（goal 模式用）、 goal_mode（新 goal 语义：1 时 prompt 即目标，无验收命令、 - 不限轮次；老行 NULL 按 0 计）。 + 不限轮次；老行 NULL 按 0 计），以及运行中改 goal 的 + goal_version、applied_goal_version（老行 NULL 按 0 计）。 PRAGMA table_info 查现有列，缺哪个补哪个；已迁移的库再跑 一次是 no-op。worker 写执行结果、server 读任务时都依赖 这些列存在。 @@ -1140,6 +1145,10 @@ def migrate_tasks_table(conn): "verify_command": "TEXT", "goal_max_rounds": "INTEGER", "goal_mode": "INTEGER", + # 运行中改 goal：goal_version 已提交版本、applied_goal_version + # 循环已应用版本（老行 NULL 按 0 计，见 _task_to_dict）。 + "goal_version": "INTEGER", + "applied_goal_version": "INTEGER", # HA 阶段 1：认领任务的 worker 所在主机名；/pr 据此判断 # worktree 在本机（同步跑）还是别机（排 pr 任务给该主机）。 "worktree_host": "TEXT", @@ -2209,6 +2218,37 @@ def _thread_non_goal_guard(task): "status": task["status"]} return None +def _goal_update_reject(task, new_goal): + """运行中改 goal 的拒绝判定（纯函数，不碰 DB）。 + + 返回 (status, error) 或 (None, None)：非新语义 goal 任务 + （goal_mode 不为 1，或 verify_command 非空）409——旧验收 + 命令形态不在本设计内，绝不静默转换（verify_argv 只活在 + host-state 里、不是任务表列，服务端以 verify_command 把关， + host 循环有验收时一律忽略信号）；已终态 409（终态走复活路）。 + 逐字相同是 no-op，不是拒绝，由调用方单独处理。 + """ + if not task.get("goal_mode") or task.get("verify_command"): + return (409, "旧验收命令形态不在运行中改 goal 设计内") + if task["status"] in ("completed", "failed", "merged", + "merge-failed", "cancelled"): + return (409, "已终态任务请走复活路，不要用运行中改 goal") + return (None, None) + + +def _goal_signal_status(status): + """改 goal 成功后，信号投递状态（仅作可见性，不冒充已应用）。 + + queued 只改库（下次认领自然读新 goal）；interrupted 等 + 续跑认领钩子写信号；running 等 worker 镜像线程写信号。 + """ + if status == "queued": + return "queued_no_signal_needed" + if status == "interrupted": + return "pending_resume_signal" + return "pending_mirror_to_host_loop" + + class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, AutomationMixin, FullAutoMixin, BaseHTTPRequestHandler): """控制面请求处理：除 /health 与 /webhooks/* 外一律要求 X-API-Key。""" @@ -2367,6 +2407,8 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au self._retry_task(user_id, rest[: -len("/retry")]) elif rest.endswith("/goal-resume") and method == "POST": self._goal_resume_task(user_id, rest[: -len("/goal-resume")]) + elif rest.endswith("/goal") and method == "POST": + self._update_goal_task(user_id, rest[: -len("/goal")]) elif rest.endswith("/auto-memory-flags") and method == "POST": self._task_auto_memory_flags( user_id, rest[: -len("/auto-memory-flags")]) @@ -7502,6 +7544,83 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au "task_goal_resumed", task_id) self._send_json(200, task) + def _update_goal_task(self, user_id, task_id): + """运行中改 goal：POST /tasks/{id}/goal。 + + 只改目标文本，不改任务状态、不建新任务、不开新分支、 + 不重置轮次/用量/预算/失败计数。生效路径：queued 只改库； + running 由 worker 镜像线程（每 30 秒）写信号，host 循环 + 在下一个轮次边界应用，绝不在正在跑的一轮中途打断； + interrupted 由续跑认领钩子写信号，断点轮次的下一轮应用。 + 拒绝口径见 _goal_update_reject；逐字相同是 no-op（不 + bump 版本、不写 history、不写审计）；改 goal 本身不花 + 模型钱，不查配额；返回体含 goal_update 与 + goal_signal_status，goal_version（已提交）与 + applied_goal_version（已应用）不等时绝不冒充已应用。 + """ + body, err = self._read_json() + if err: + self._send_json(400, {"error": err}) + return + if not isinstance(body, dict): + self._send_json(400, {"error": "body must be a JSON object"}) + return + new_goal = body.get("goal") + if not isinstance(new_goal, str) or not new_goal.strip(): + self._send_json(400, {"error": "goal 不能为空"}) + return + 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"], "approve_merge"): + self._send_json(403, {"error": "forbidden"}) + return + reject_status, reject_error = _goal_update_reject(task, new_goal) + if reject_status is not None: + self._send_json(reject_status, { + "error": reject_error, "status": task["status"]}) + return + current_goal = task.get("goal") or task.get("prompt") or "" + if new_goal == current_goal: + self._send_json(200, dict( + task, goal_update="noop", + goal_signal_status="noop_same_goal")) + return + now = time.time() + conn = self._db() + try: + migrate_tasks_table(conn) + _glu_ensure_history(conn) + new_version = _glu_apply_goal_update( + conn, task_id, current_goal, + task.get("goal_version") or 0, new_goal, user_id, now) + if new_version is None: + conn.rollback() + self._send_json(409, { + "error": "任务已变化，请刷新后重试", + "status": task["status"], + }) + return + conn.commit() + task = _task_to_dict( + 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_goal_updated", + "%s:goal_version=%s" % (task_id, new_version)) + self._send_json(200, dict( + task, goal_update="updated", + goal_signal_status=_goal_signal_status(task["status"]))) + def _retry_task(self, user_id, task_id): """把终态失败任务重新排队：只改状态，不直接执行。 diff --git a/cloud/tests/test_goal_live_update.py b/cloud/tests/test_goal_live_update.py new file mode 100644 index 0000000..dba0af3 --- /dev/null +++ b/cloud/tests/test_goal_live_update.py @@ -0,0 +1,509 @@ +"""运行中改 goal：端点、信号格式、worker 桥、host 循环。 + +只用假 turn / 假 state / 临时 SQLite，真实模型 0 轮，不联网， +不碰生产。host 循环的假 turn 直接调 host.run_task_turn 的 +桩，host.py 本身可 import（顶层只有定义 + __main__ 守卫）。 + +覆盖任务书逐项：403、空 goal 400、相同 no-op（版本与 +history 不变）、终态/merged 409、旧验收形态 409、queued +只改库、运行中信号在轮次边界生效且轮次/用量/失败计数不 +重置、旧宣告作废（新 goal 下需重新宣告）、过期版本被丢弃、 +最新版胜出、已中断续跑带新 goal 且轮次不断点。 +""" + +import contextlib +import json +import os +import sqlite3 +import sys +import tempfile +import time +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT)) + +import host as host_mod # noqa: E402 +from cloud import goal_live_update as glu # noqa: E402 +from cloud import server as lifecycle_mod # noqa: E402 +# Reconcile-retarget: _update_goal_task lives in cloud.server on this +# line (target keeps the 6-mixin split); the test patches _auth_store +# in the module where the method resolves it, i.e. cloud.server. +from cloud.server import CloudControlPlaneHandler, init_db # noqa: E402 +from cloud.worker import Worker # noqa: E402 + + +def _handler(db_path, body, permitted=True): + h = CloudControlPlaneHandler.__new__(CloudControlPlaneHandler) + h.server = SimpleNamespace(db_path=db_path) + sent = [] + h._send_json = lambda status, obj, extra_headers=None: sent.append( + (status, obj)) + h._read_json = lambda: (body, None) + h.sent = sent + audits = [] + + class FakeStore: + def check_permission(self, user_id, project_id, action): + return permitted and action == "approve_merge" + + def audit(self, user_id, project_id, action, detail=""): + audits.append((user_id, project_id, action, detail)) + + @contextlib.contextmanager + def fake_auth_store(_db_path): + yield FakeStore() + + h._audits = audits + return h, fake_auth_store + + +def _insert_task(db_path, task_id="T1", status="running", + goal="old goal", goal_mode=1, verify_command=None, + goal_version=0, applied=0): + now = time.time() + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " idempotency_key, depends_on, priority, created_by," + " created_at, updated_at, last_heartbeat, goal, goal_mode," + " verify_command, goal_max_rounds, goal_version," + " applied_goal_version) VALUES(?, 'p1', ?, ?, NULL, '[]'," + " 0, 'tester', ?, ?, NULL, ?, ?, ?, 0, ?, ?)", + (task_id, goal, status, now, now, goal, goal_mode, + verify_command, goal_version, applied)) + conn.commit() + finally: + conn.close() + + +def _task_row(db_path, task_id="T1"): + conn = sqlite3.connect(db_path) + try: + conn.row_factory = sqlite3.Row + return dict(conn.execute( + "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()) + finally: + conn.close() + + +def _history(db_path, task_id="T1"): + conn = sqlite3.connect(db_path) + try: + return conn.execute( + "SELECT old_goal, new_goal, version, created_by FROM" + " task_goal_history WHERE task_id = ? ORDER BY version", + (task_id,)).fetchall() + finally: + conn.close() + + +class SignalFormatTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.state_path = os.path.join(self.tmp.name, "host-state.json") + with open(self.state_path, "w", encoding="utf-8") as fh: + json.dump({"tasks": {}}, fh) + self.host_name = "codex-abc123" + + def tearDown(self): + self.tmp.cleanup() + + def test_cloud_and_host_signal_paths_agree(self): + self.assertEqual( + glu.signal_path(self.state_path, self.host_name), + host_mod._goal_update_signal_path( + self.state_path, self.host_name)) + + def test_write_read_latest_wins_and_stale_dropped(self): + glu.write_goal_update_signal( + self.state_path, self.host_name, 2, "goal v2", "T1") + # 版本 3 覆盖版本 2：只应用最新版，中间版本不进循环。 + glu.write_goal_update_signal( + self.state_path, self.host_name, 3, "goal v3", "T1") + self.assertEqual( + glu.read_goal_update_signal(self.state_path, self.host_name, 1), + (3, "goal v3")) + # 已应用版本 3 后，同一文件即过期。 + self.assertIsNone( + glu.read_goal_update_signal(self.state_path, self.host_name, 3)) + # host 读侧同款判定。 + self.assertIsNone( + host_mod._read_goal_update_signal( + self.state_path, self.host_name, 3)) + + def test_bad_signal_never_raises(self): + path = glu.signal_path(self.state_path, self.host_name) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fh: + fh.write("{not json") + self.assertIsNone( + glu.read_goal_update_signal(self.state_path, self.host_name, 0)) + with open(path, "w", encoding="utf-8") as fh: + json.dump({"version": 5, "goal": " "}, fh) + self.assertIsNone( + glu.read_goal_update_signal(self.state_path, self.host_name, 0)) + with open(path, "w", encoding="utf-8") as fh: + json.dump({"version": "5", "goal": "x"}, fh) + self.assertIsNone( + glu.read_goal_update_signal(self.state_path, self.host_name, 0)) + + def test_consume_only_deletes_applied_or_older(self): + glu.write_goal_update_signal( + self.state_path, self.host_name, 4, "goal v4", "T1") + self.assertFalse(glu.consume_goal_update_signal( + self.state_path, self.host_name, 3)) + self.assertTrue(os.path.exists( + glu.signal_path(self.state_path, self.host_name))) + self.assertTrue(glu.consume_goal_update_signal( + self.state_path, self.host_name, 4)) + self.assertFalse(os.path.exists( + glu.signal_path(self.state_path, self.host_name))) + + def test_notice_mentions_old_new_and_void(self): + notice = glu.build_goal_update_notice("old goal", "new goal") + self.assertIn("old goal", notice) + self.assertIn("new goal", notice) + self.assertIn("作废", notice) + self.assertEqual( + notice, host_mod._goal_update_notice("old goal", "new goal")) + + +class EndpointTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "test.db") + init_db(self.db_path) + + def tearDown(self): + self.tmp.cleanup() + + def _call(self, task_id, body, permitted=True): + h, fake_auth = _handler(self.db_path, body, permitted) + with mock.patch.object(lifecycle_mod, "_auth_store", fake_auth): + h._update_goal_task("tester", task_id) + return h.sent[-1] + + def test_forbidden_without_approve_merge(self): + _insert_task(self.db_path) + status, _ = self._call("T1", {"goal": "new goal"}, + permitted=False) + self.assertEqual(status, 403) + self.assertEqual(_task_row(self.db_path)["goal"], "old goal") + + def test_empty_goal_400(self): + _insert_task(self.db_path) + status, _ = self._call("T1", {"goal": " "}) + self.assertEqual(status, 400) + + def test_same_goal_noop_changes_nothing(self): + _insert_task(self.db_path, goal_version=2, applied=2) + status, obj = self._call("T1", {"goal": "old goal"}) + self.assertEqual(status, 200) + self.assertEqual(obj["goal_update"], "noop") + row = _task_row(self.db_path) + self.assertEqual(row["goal_version"], 2) + # history 表由改 goal 端点建表，no-op 路不建表也不写行： + # 先建表再断言为空，才是“no-op 没写 history”的证据。 + conn = sqlite3.connect(self.db_path) + try: + glu.ensure_goal_history_table(conn) + conn.commit() + finally: + conn.close() + self.assertEqual(_history(self.db_path), []) + + def test_terminal_and_merged_409(self): + for status_name in ("completed", "failed", "merged", + "merge-failed", "cancelled"): + _insert_task(self.db_path, task_id="X-" + status_name, + status=status_name) + status, _ = self._call("X-" + status_name, + {"goal": "new goal"}) + self.assertEqual(status, 409, status_name) + + def test_old_verify_form_409(self): + _insert_task(self.db_path, verify_command="make check") + status, _ = self._call("T1", {"goal": "new goal"}) + self.assertEqual(status, 409) + _insert_task(self.db_path, task_id="T2", goal_mode=0) + status, _ = self._call("T2", {"goal": "new goal"}) + self.assertEqual(status, 409) + + def test_queued_only_updates_db_and_history(self): + _insert_task(self.db_path, status="queued") + status, obj = self._call("T1", {"goal": "new goal"}) + self.assertEqual(status, 200) + self.assertEqual(obj["goal_update"], "updated") + self.assertEqual(obj["goal_signal_status"], + "queued_no_signal_needed") + row = _task_row(self.db_path) + self.assertEqual(row["goal"], "new goal") + self.assertEqual(row["goal_version"], 1) + self.assertEqual(row["applied_goal_version"], 0) + self.assertEqual(_history(self.db_path), + [("old goal", "new goal", 1, "tester")]) + + def test_running_and_interrupted_signal_status(self): + _insert_task(self.db_path, status="running") + _, obj = self._call("T1", {"goal": "new goal"}) + self.assertEqual(obj["goal_signal_status"], + "pending_mirror_to_host_loop") + _insert_task(self.db_path, task_id="T2", status="interrupted") + _, obj = self._call("T2", {"goal": "new goal"}) + self.assertEqual(obj["goal_signal_status"], + "pending_resume_signal") + + def test_missing_task_404(self): + status, _ = self._call("NOPE", {"goal": "new goal"}) + self.assertEqual(status, 404) + + +class WorkerBridgeTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "test.db") + init_db(self.db_path) + self.state_dir = os.path.join(self.tmp.name, "work") + os.makedirs(self.state_dir, exist_ok=True) + self.state_path = os.path.join(self.state_dir, "host-state.json") + self.task = {"id": "T1", "project_id": "p1"} + self.host_name = "codex-abc123" + + def tearDown(self): + self.tmp.cleanup() + + def _worker(self): + w = Worker.__new__(Worker) + w.db_path = self.db_path + w.tenant_isolation = False + w.tenants_root = os.path.join(self.tmp.name, "tenants") + executor = SimpleNamespace( + extra_mounts=[(self.state_dir, "/app/work")], + config_path=None) + executor._host_name = lambda task: self.host_name + w.executor = executor + return w + + def _write_state(self, applied): + with open(self.state_path, "w", encoding="utf-8") as fh: + json.dump({"tasks": {self.host_name: { + "name": self.host_name, "thread_id": "th1", + "goal": "old goal", "goal_rounds_done": 2, + "applied_goal_version": applied, + "status": "goal-running"}}}, fh) + + def test_resume_hook_writes_signal_when_db_newer(self): + _insert_task(self.db_path, status="interrupted", + goal="new goal", goal_version=3) + self._write_state(applied=1) + w = self._worker() + self.assertTrue(w._sync_resume_goal_signal(self.task)) + self.assertEqual( + glu.read_goal_update_signal( + self.state_path, self.host_name, 1), (3, "new goal")) + + def test_resume_hook_no_signal_when_not_newer(self): + _insert_task(self.db_path, status="interrupted", + goal="old goal", goal_version=1) + self._write_state(applied=1) + w = self._worker() + self.assertFalse(w._sync_resume_goal_signal(self.task)) + self.assertIsNone(glu.read_goal_update_signal( + self.state_path, self.host_name, 1)) + + def test_mirror_writes_signal_and_reports_applied(self): + _insert_task(self.db_path, status="running", + goal="new goal", goal_version=2) + self._write_state(applied=1) + w = self._worker() + rounds = w._mirror_checkpoint_once(self.task) + self.assertEqual(rounds, 2) + self.assertEqual( + glu.read_goal_update_signal( + self.state_path, self.host_name, 1), (2, "new goal")) + row = _task_row(self.db_path) + self.assertEqual(row["applied_goal_version"], 1) + + +def _fake_turn_script(prompts, script, signal_writer=None): + """返回 host.run_task_turn 的桩：按 script 顺序回回合。 + + script 每项是 (status, text, usage)；signal_writer 若给， + 在第 signal_writer[0] 轮执行时先写信号再回回合（模拟 + 本轮进行中提交新 goal，中途不打断本轮）。 + """ + state = {"n": 0} + + def fake(server, st, task, prompt, model, effort, sandbox, + timeout=None, interrupt_file=None): + prompts.append(prompt) + idx = state["n"] + state["n"] += 1 + if signal_writer is not None and idx == signal_writer[0]: + signal_writer[1]() + status, text, usage = script[idx] + return {"status": status, "text": text, "usage": usage} + + return fake + + +class HostLoopTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.state_path = os.path.join(self.tmp.name, "host-state.json") + self.state = {"tasks": {}} + + def tearDown(self): + self.tmp.cleanup() + + def _task(self, name="codex-abc123"): + task = {"name": name, "cwd": self.tmp.name, "thread_id": "th1", + "turns": []} + self.state["tasks"][name] = task + return task + + def _write_signal(self, version, goal): + glu.write_goal_update_signal( + self.state_path, "codex-abc123", version, goal, "T1") + + def test_mid_turn_update_applies_at_next_boundary(self): + task = self._task() + prompts = [] + script = [ + ("completed", "did old work", + {"input_tokens": 10, "output_tokens": 1, "total_tokens": 11, + "cached_input_tokens": 0, "cache_write_input_tokens": 0, + "reasoning_output_tokens": 0, "completions": 1}), + ("completed", "目标已达成", + {"input_tokens": 20, "output_tokens": 1, "total_tokens": 21, + "cached_input_tokens": 0, "cache_write_input_tokens": 0, + "reasoning_output_tokens": 0, "completions": 1}), + ] + fake = _fake_turn_script( + prompts, script, + signal_writer=(0, lambda: self._write_signal(1, "new goal"))) + with mock.patch.object(host_mod, "run_task_turn", fake): + result = host_mod.run_goal_loop( + SimpleNamespace(), self.state, self.state_path, task, + "old goal", None, 0, None, None, None, cfg=None) + self.assertTrue(result["ok"]) + self.assertEqual(result["rounds_done"], 2) + # 第一轮提示仍是旧 goal（中途不打断），第二轮才是新 goal。 + self.assertIn("old goal", prompts[0]) + self.assertNotIn("new goal", prompts[0]) + self.assertIn("new goal", prompts[1]) + self.assertIn("old goal", prompts[1]) + self.assertIn("作废", prompts[1]) + # 用量累加、不重置；state 检查点记新 goal 与已应用版本。 + self.assertEqual(result["usage"]["input_tokens"], 30) + saved = json.load(open(self.state_path, encoding="utf-8")) + entry = saved["tasks"]["codex-abc123"] + self.assertEqual(entry["goal"], "new goal") + self.assertEqual(entry["applied_goal_version"], 1) + self.assertEqual(entry["goal_rounds_done"], 2) + + def test_old_declaration_void_when_goal_changes_mid_turn(self): + task = self._task() + prompts = [] + script = [ + ("completed", "做完了\n目标已达成", + {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2, + "cached_input_tokens": 0, "cache_write_input_tokens": 0, + "reasoning_output_tokens": 0, "completions": 1}), + ("completed", "新目标做完了\n目标已达成", + {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2, + "cached_input_tokens": 0, "cache_write_input_tokens": 0, + "reasoning_output_tokens": 0, "completions": 1}), + ] + fake = _fake_turn_script( + prompts, script, + signal_writer=(0, lambda: self._write_signal(1, "new goal"))) + with mock.patch.object(host_mod, "run_task_turn", fake): + result = host_mod.run_goal_loop( + SimpleNamespace(), self.state, self.state_path, task, + "old goal", None, 0, None, None, None, cfg=None) + # 旧 goal 的宣告不能结束新 goal 的循环，必须重新宣告。 + self.assertTrue(result["ok"]) + self.assertEqual(result["rounds_done"], 2) + self.assertIn("new goal", prompts[1]) + + def test_failed_counter_not_reset_by_goal_change(self): + task = self._task() + prompts = [] + script = [ + ("failed", "boom 1", None), + ("failed", "boom 2", None), + ("failed", "boom 3", None), + ] + fake = _fake_turn_script( + prompts, script, + signal_writer=(1, lambda: self._write_signal(1, "new goal"))) + with mock.patch.object(host_mod, "run_task_turn", fake): + result = host_mod.run_goal_loop( + SimpleNamespace(), self.state, self.state_path, task, + "old goal", None, 0, None, None, None, cfg=None) + # 改 goal 不能绕过连续 3 轮失败护栏：第 3 轮失败仍中断。 + # rounds_done 是既有语义：连续失败中断在检查点落盘前 + # 返回，所以第 3 轮失败不记入 rounds_done（停在 2）， + # 本批未改该语义；关键证据是第 3 轮确实跑了且已用新 goal。 + self.assertTrue(result["interrupted"]) + self.assertEqual(len(prompts), 3) + self.assertEqual(result["rounds_done"], 2) + # 第三轮提示已是新 goal（第二轮结束后应用）。 + self.assertIn("new goal", prompts[2]) + + def test_verify_loop_ignores_signal(self): + task = self._task() + self._write_signal(1, "new goal") + prompts = [] + script = [("completed", "done", None)] + fake = _fake_turn_script(prompts, script) + with mock.patch.object(host_mod, "run_task_turn", fake), \ + mock.patch.object( + host_mod, "run_gated_verify", + lambda *a, **k: (0, "ok")): + result = host_mod.run_goal_loop( + SimpleNamespace(), self.state, self.state_path, task, + "old goal", "make check", 3, None, None, None, cfg=None) + self.assertTrue(result["ok"]) + self.assertNotIn("目标已更新", prompts[0]) + saved = json.load(open(self.state_path, encoding="utf-8")) + entry = saved["tasks"]["codex-abc123"] + self.assertEqual(entry["goal"], "old goal") + self.assertEqual(entry["applied_goal_version"], 0) + + def test_interrupted_resume_carries_new_goal_without_reset(self): + # 断点续跑：state 里轮次已到 2，认领钩子写好的信号在 + # 第一个续跑轮次（第 3 轮）边界应用，轮次不回第 1 轮。 + task = self._task() + task["goal"] = "old goal" + task["goal_rounds_done"] = 2 + task["applied_goal_version"] = 0 + task["status"] = "interrupted" + host_mod.save_state(self.state_path, self.state) + self._write_signal(1, "new goal") + prompts = [] + script = [ + ("completed", "继续做", None), + ("completed", "新目标做完了\n目标已达成", None), + ] + fake = _fake_turn_script(prompts, script) + with mock.patch.object(host_mod, "run_task_turn", fake): + result = host_mod.run_goal_loop( + SimpleNamespace(), self.state, self.state_path, task, + "old goal", None, 0, None, None, None, cfg=None, + start_round=3, last_verify="", last_turn_status=None, + last_turn_text="") + self.assertTrue(result["ok"]) + self.assertEqual(result["rounds_done"], 4) + self.assertIn("new goal", prompts[0]) + self.assertIn("旧目标", prompts[0]) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 3373a9f..302469c 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -168,6 +168,9 @@ from cloud.sandbox_policy import cleanup_decision as _cleanup_decision from cloud.timeouts import STALE_SWEEP_SECONDS as _STALE_SWEEP from cloud.db import connect as _db_connect from cloud.goal_command import parse_goal_command as _parse_goal_command +from cloud.goal_live_update import ( + write_goal_update_signal as _glu_write_signal, +) from cloud.logging_config import get_logger, log_event, setup_logging from cloud.models import ModelRouter, is_fallbackable, record_fallback from cloud.auto_memory import AutoMemoryStore @@ -433,6 +436,10 @@ def _task_from_row(row): # 新 goal 语义：goal_mode=1 时 prompt 即目标，无验收命令、 # 不限轮次；老行缺列时 None = 按 verify_command 判定。 "goal_mode": _col("goal_mode"), + # 运行中改 goal：已提交版本与循环已应用版本，老行缺列 + # 时按 0 计（= 从未改过 goal，信号版本从 1 起算）。 + "goal_version": _col("goal_version") or 0, + "applied_goal_version": _col("applied_goal_version") or 0, # 任务级选项：老行缺列时 None = 跟随配置（见 server 侧）。 "model": _col("model"), "effort": _col("effort"), @@ -1270,6 +1277,68 @@ class Worker: except Exception: return None + def _write_goal_signal_if_newer(self, task, db_goal, db_version): + """运行中改 goal 的信号写侧（worker 侧唯一写点）。 + + state 文件里的 applied_goal_version 是 host 循环上次 + 检查点落盘的已应用版本；库里 goal_version 更大时， + 把新 goal 写成信号文件（host-state 同目录的 + goal-updates/.json，host 循环在轮次边界读）。 + 任何异常一律返回 False，信号写失败绝不能炸掉任务 + 执行或镜像本身。state 文件不存在（Local/Docker 的 + work 挂载还没建起来、或 CloudExecutor 根本没有 + host-state）时同样返回 False，不猜路径。 + """ + try: + if not db_goal or not isinstance(db_goal, str): + return False + version = int(db_version or 0) + if version < 1: + return False + state_path = self._host_state_path_for_task(task) + if not state_path or not os.path.exists(state_path): + return False + with open(state_path, encoding="utf-8") as fh: + state = json.load(fh) + host_name = _worker_host_name(self.executor, task) + entry = (state.get("tasks") or {}).get(host_name) + if not isinstance(entry, dict): + return False + applied = int(entry.get("applied_goal_version") or 0) + if version <= applied: + return False + _glu_write_signal(state_path, host_name, version, + db_goal, task.get("id")) + return True + except Exception: + return False + + def _sync_resume_goal_signal(self, task): + """断点续跑认领时钩子：库里新 goal 先写成信号再起循环。 + + 镜像线程只在循环跑起来之后才工作，已中断任务没有 + 在跑的镜像线程，所以认领断点续跑时必须先读一次库： + 库里 goal_version 大于 state 文件已应用版本时写信号， + host 循环在断点轮次的下一轮（第一个轮次边界）应用 + 它，轮次不断点、不回到第 1 轮。任何异常返回 False， + 续跑本身照常走断点（用旧 goal），不因信号失败而炸。 + """ + try: + conn = _db_connect(self.db_path) + try: + row = conn.execute( + "SELECT goal, goal_version FROM tasks WHERE id = ?", + (task["id"],)).fetchone() + finally: + conn.close() + if row is None: + return False + goal = row[0] if len(row) > 0 else None + version = row[1] if len(row) > 1 else None + return self._write_goal_signal_if_newer(task, goal, version) + except Exception: + return False + def _mirror_checkpoint_once(self, task): """把磁盘 host-state 里的本任务断点镜像进库（单次）。 @@ -1302,6 +1371,34 @@ class Worker: conn = _db_connect(self.db_path) try: self._ensure_result_columns(conn) + # 运行中改 goal 可见性（只在变化时写）：host 循环 + # 检查点里的 applied_goal_version 回写任务行， + # 调用方据此区分“已提交的新版本”与“循环已应用 + # 的版本”，两者不等时绝不冒充已应用。 + if entry.get("applied_goal_version") is not None: + try: + applied_v = int(entry["applied_goal_version"]) + except (TypeError, ValueError): + applied_v = None + if applied_v is not None: + conn.execute( + "UPDATE tasks SET applied_goal_version = ?" + " WHERE id = ? AND (applied_goal_version" + " IS NULL OR applied_goal_version <> ?)", + (applied_v, task["id"], applied_v)) + # 运行中改 goal 反向信号：库里 goal_version 大于 + # state 已应用版本时写信号文件，host 循环在下个 + # 轮次边界应用；读库失败时跳过（下个 30 秒周期 + # 再试），信号写失败由 helper 内部吞掉。 + try: + db_row = conn.execute( + "SELECT goal, goal_version FROM tasks" + " WHERE id = ?", (task["id"],)).fetchone() + except Exception: + db_row = None + if db_row is not None: + self._write_goal_signal_if_newer( + task, db_row[0], db_row[1]) existing = conn.execute( "SELECT 1 FROM task_checkpoints" " WHERE task_id = ? AND round_no = ?", @@ -1525,6 +1622,9 @@ class Worker: "verify_command": "TEXT", "goal_max_rounds": "INTEGER", "goal_mode": "INTEGER", + # 运行中改 goal：与 server.migrate_tasks_table 同列。 + "goal_version": "INTEGER", + "applied_goal_version": "INTEGER", "worktree_host": "TEXT", "claimed_by_worker": "TEXT", "model": "TEXT", @@ -4688,6 +4788,15 @@ class Worker: except Exception: _mirror_stop = None _mirror_thread = None + # 运行中改 goal 认领钩子：断点续跑时镜像线程还没跑， + # 必须在起循环之前先把库里新 goal 写成信号，否则断点 + # 续跑会沿用 host-state 里的旧 goal。钩子只写信号， + # 失败时续跑照常用旧 goal，不炸执行（见 helper 文档）。 + if goal_mode and task.get("_goal_resume_breakpoint"): + try: + self._sync_resume_goal_signal(task) + except Exception: + pass if is_docker and hasattr(self.executor, "kill_container"): try: _container_name = self.executor.container_name(task) diff --git a/host.py b/host.py index fce2260..149083c 100644 --- a/host.py +++ b/host.py @@ -129,6 +129,95 @@ def _interrupt_request_path(state_path, task_name): "interrupt-requests", safe + ".json") +def _goal_update_signal_path(state_path, task_name): + """运行中改 goal 的信号文件路径（与 cloud/goal_live_update.py + 同款格式，改一处必须同步另一处，跨模块测试钉住它）： + /goal-updates/.json，内容是 + {"version": int, "goal": str, "task_id": str | None}。 + 只由 worker 侧（镜像线程 / 断点认领钩子）写，host 循环 + 只在轮次边界读，绝不在正在跑的一轮中途读它。 + """ + safe = re.sub(r"[^A-Za-z0-9_.-]", "-", str(task_name)) + return os.path.join( + os.path.dirname(os.path.abspath(state_path)), + "goal-updates", safe + ".json") + + +def _read_goal_update_signal(state_path, task_name, applied_version): + """读信号，返回 (version, goal) 或 None。 + + 文件不存在、JSON 坏、version 不是正整数、goal 不是非空 + 字符串、version <= 已应用版本时一律返回 None，绝不抛。 + """ + try: + applied = int(applied_version or 0) + except (TypeError, ValueError): + applied = 0 + try: + with open(_goal_update_signal_path(state_path, task_name), + encoding="utf-8") as fh: + data = json.load(fh) + except Exception: + return None + if not isinstance(data, dict): + return None + version = data.get("version") + goal = data.get("goal") + if not isinstance(version, int) or isinstance(version, bool) \ + or version < 1: + return None + if not isinstance(goal, str) or not goal.strip(): + return None + if version <= applied: + return None + return (version, goal) + + +def _consume_goal_update_signal(state_path, task_name, applied_version): + """已应用版本的信号文件尽力删掉（删失败静默）。 + + 只删 version <= 已应用版本的信号；更新的信号（写于应用 + 之后、尚未应用）绝不能被误删。 + """ + try: + applied = int(applied_version or 0) + except (TypeError, ValueError): + applied = 0 + path = _goal_update_signal_path(state_path, task_name) + try: + with open(path, encoding="utf-8") as fh: + data = json.load(fh) + except Exception: + return False + if not isinstance(data, dict): + return False + version = data.get("version") + if not isinstance(version, int) or isinstance(version, bool): + return False + if version > applied: + return False + try: + os.remove(path) + except Exception: + return False + return True + + +def _goal_update_notice(old_goal, new_goal): + """轮次边界应用新 goal 时喂给 agent 的显式通知块（与 + cloud/goal_live_update.build_goal_update_notice 同款文案）。 + """ + return ( + "目标已更新（运行中改 goal）：\n" + "旧目标：%s\n" + "新目标：%s\n" + "请继续完成新目标，不要重复已完成的部分；完成与否只按" + "新目标判断；此前如已输出过“目标已达成”，该宣告对新" + "目标作废——新目标满足时请重新宣告，否则继续工作。" + % (old_goal, new_goal) + ) + + def _wait_turn(q, timeout, progress_interval=30, on_progress=None, interrupt_check=None): """等 turn/completed 的轮询等待（对齐 Codex App 的长等待语义）。 @@ -2680,6 +2769,15 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify, task["goal_max_rounds"] = int(max_rounds) task["goal_rounds_done"] = int(start_round) - 1 task["goal_last_verify"] = last_verify + # 运行中改 goal：已应用版本只活在 state 检查点里（断点 + # 续跑时从 state 条目恢复），老条目缺字段时按 0 计；新 + # 循环的 spawn goal 本身就是版本 0 的起点，信号版本从 1 起。 + try: + applied_goal_version = int(task.get("applied_goal_version") or 0) + except (TypeError, ValueError): + applied_goal_version = 0 + task["applied_goal_version"] = applied_goal_version + pending_goal_notice = None # 循环前先落盘一次：否则第一轮 turn 进行中被杀时，磁盘上的 # 任务还没有 goal/verify_command，goal-resume 会因“没有记录” # 而拒绝——断点续跑必须在第一轮开始前就已可续跑。 @@ -2710,6 +2808,38 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify, "rounds_done": task["goal_rounds_done"], "last_verify": last_verify, "usage": goal_usage} + def _apply_goal_update(): + """运行中改 goal 的轮次边界应用（只换内存里的 goal）。 + + 有验收的循环一律忽略信号（旧验收命令形态不在运行中 + 改 goal 设计内，服务端已用 verify_command 把关，argv + 形态只活在 host-state 里，这里是第二道门）。信号版本 + 大于已应用版本时才换：goal_text、task["goal"]、 + applied_goal_version 一起换，并留一条待用通知给本轮 + 或下一轮的提示。落盘由本轮检查点做（换后、检查点前 + 被杀时 state 里仍是旧版本、信号文件还在，断点续跑 + 会重新应用，绝不丢更新）；轮次计数、用量、连续失败 + 计数、预算状态一律不动。 + """ + nonlocal goal_text, applied_goal_version, pending_goal_notice + if has_verify: + return False + task_name = task.get("name") + if not task_name: + return False + sig = _read_goal_update_signal( + state_path, task_name, applied_goal_version) + if sig is None: + return False + version, new_goal = sig + old_goal = goal_text + goal_text = new_goal + applied_goal_version = version + task["goal"] = new_goal + task["applied_goal_version"] = version + pending_goal_notice = _goal_update_notice(old_goal, new_goal) + return True + # max_rounds=0 是不限轮次：while 循环只在模型宣告、预算耗尽 # 或取消时结束；非 0 时按旧语义穷尽轮次即 unmet。 round_no = int(start_round) - 1 @@ -2728,6 +2858,10 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify, print(f"goal stopped: 预算已用完（max_daily_turns），" f"第 {round_no} 轮未开始，任务可续跑") break + # 运行中改 goal 检查点一：轮次开始前（构造本轮提示 + # 之前）应用信号。正在跑的一轮绝不在中途被打断， + # 信号只影响从本轮起的提示。 + _apply_goal_update() if round_no == 1 and not last_verify and not last_turn_status: prompt = goal_text if not has_verify: @@ -2766,6 +2900,11 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify, f"目标：{goal_text}\n" f"验收命令：{verify_text}\n" f"上一轮验收输出：\n{last_verify}") + if pending_goal_notice is not None: + # 本轮提示显式写出旧 goal → 新 goal；通知只用一次， + # 下一轮提示是重新构造的，不重复贴。 + prompt = pending_goal_notice + "\n\n" + prompt + pending_goal_notice = None try: result = run_task_turn(server, state, task, prompt, model, effort, sandbox, timeout=timeout, @@ -2829,6 +2968,12 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify, save_state(state_path, state) except Exception as exc: return _interrupt_loop(exc) + # 已应用版本的信号文件尽力清掉；更新的信号（写于本 + # 轮检查点之后、尚未应用）版本号大于已应用版本， + # 清理函数不会误删它。 + if task.get("name"): + _consume_goal_update_signal( + state_path, task["name"], applied_goal_version) print(f"round {round_no}: turn={result['status']} " f"verify_rc={v_rc if v_rc is not None else '-'} " f"verify_out={last_verify[:300]}") @@ -2851,9 +2996,17 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify, ensure_ascii=False) + "\n") except Exception: pass + # 运行中改 goal 检查点二：回合结束后、宣告判定之前 + # 再查一次信号。本轮进行中提交的新 goal 会在这里 + # 应用：本轮对旧 goal 的宣告作废，不计入新 goal 的 + # 完成判定，循环继续到下一轮用新 goal 提示；轮次计数、 + # 用量、连续失败计数一律不动。有验收的循环不应用 + # 信号（_apply_goal_update 内部已挡），判定原样。 + goal_updated_after_turn = _apply_goal_update() if not has_verify: # 模型自决：只有完整跑完的回合里的标记行才算数。 - if (result["status"] == "completed" + if (not goal_updated_after_turn + and result["status"] == "completed" and _llm_declared_done(last_turn_text)): ok = True break @@ -4018,6 +4171,14 @@ def main(): args.approval_policy, seed_dir=args.seed_dir) save_state(args.state, state) + # 新开循环的 spawn goal 是权威的：上一轮同名循环 + # 残留的信号文件必须清掉，否则首轮边界会把它当 + # 新信号应用、覆盖刚创建的 goal。循环跑起来之后 + # 镜像线程会按需重写信号，清早期的不丢更新。 + try: + os.remove(_goal_update_signal_path(args.state, args.name)) + except Exception: + pass if args.verify and args.verify_argv: print("verify_argv 已指定，--verify 字符串被忽略") run_goal_loop(srv, state, args.state, task, args.task, @@ -4061,6 +4222,16 @@ def main(): return max_rounds = args.max_rounds \ if args.max_rounds is not None else 0 + # 开新目标的 --goal-text 是权威的：残留的运行中 + # 改 goal 信号必须清掉，否则首轮边界会把它当新 + # 信号应用、覆盖这次明确传入的新目标。断点续跑 + # 分支（不传 --goal-text）绝不能清——信号正是 + # 已中断任务改 goal 后续跑的投递通道。 + try: + os.remove(_goal_update_signal_path( + args.state, args.task)) + except Exception: + pass srv.resume_thread(t["thread_id"]) run_goal_loop(srv, state, args.state, t, args.goal_text, None, max_rounds,