code-to10 goal live update: POST /tasks/{id}/goal updates a running goal loop at the next round boundary
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.
This commit is contained in:
host committed
1 parent
51b26633b8
commit
b3f3f8bcab
6 files changed
+1114
-2
No files matched your search
@@ -0,0 +1,195 @@
|
||||
"""运行中改 goal 的纯函数与信号文件格式。
|
||||
|
||||
信号文件格式（与 host.py 内联读侧同款契约，改一处必须同步
|
||||
另一处，跨模块测试钉住它）：
|
||||
<state_dir>/goal-updates/<safe_host_name>.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
|
||||
Reference in new issue
Block a user