diff --git a/cloud/tests/test_goal_resume.py b/cloud/tests/test_goal_resume.py index cc133df..fbe9282 100644 --- a/cloud/tests/test_goal_resume.py +++ b/cloud/tests/test_goal_resume.py @@ -170,6 +170,20 @@ class GoalResumeTestCase(unittest.TestCase): "POST", "/tasks/%s/goal-resume" % task["id"], {}, self.owner_key) self.assertEqual(status, 200, body) + # 本测试测的是“有断点的任务走 goal-resume”：先给任务补 + # 一行库断点（模拟跑过一轮），否则按从未启动回落规则， + # 它会被当成从未启动而走全新 goal（见 + # test_neverstart_fresh.py）。 + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO task_checkpoints(task_id, round_no," + " thread_id, goal_status, state_sha, verify_out," + " created_at) VALUES(?, ?, ?, ?, ?, ?, ?)", + (task["id"], 1, "th-test", "goal-running", "sha", "", 0)) + conn.commit() + finally: + conn.close() seen = [] def fake(cmd): diff --git a/cloud/tests/test_neverstart_fresh.py b/cloud/tests/test_neverstart_fresh.py new file mode 100644 index 0000000..eec1d18 --- /dev/null +++ b/cloud/tests/test_neverstart_fresh.py @@ -0,0 +1,143 @@ +"""自动续跑从未启动回落全新跑（2026-10-29 实测任务 A）。 + +任务 A 认领后 61 秒 interrupted，但从未落盘（无 state 条目、 +无 worktree、无 trace、无库断点）；自动续跑走 goal-resume +被 host 拒绝 → failed。拒绝只对“有进度但断点丢了”有意义， +从未启动的任务唯一有意义的结果是全新跑。 + +判据四者全无才算从未启动，任一存在都不回落；查不出来时 +fail-closed，同样不回落。 +""" + +import json +import os +import sqlite3 +import sys +import tempfile +import unittest +from pathlib import Path + +try: + from cloud.sandbox import LocalExecutor + from cloud.server import init_db + from cloud.worker import Worker +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.sandbox import LocalExecutor + from cloud.server import init_db + from cloud.worker import Worker + + +def _insert_task(db_path, task_id, **cols): + import time + 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)" + " VALUES(?, ?, ?, ?, NULL, '[]', 0, ?, ?, ?, NULL)", + (task_id, "p1", "do work", "interrupted", "tester", now, now)) + for key, value in cols.items(): + conn.execute( + "UPDATE tasks SET %s = ? WHERE id = ?" % key, + (value, task_id)) + conn.commit() + finally: + conn.close() + + +class NeverStartedTestCase(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) + cfg_path = os.path.join(self.tmp.name, "host-config.json") + with open(cfg_path, "w", encoding="utf-8") as fh: + json.dump({"state": "work/host-state.json"}, fh) + self.state_path = os.path.join(self.state_dir, "host-state.json") + with open(self.state_path, "w", encoding="utf-8") as fh: + json.dump({"version": 1, "tasks": {}}, fh) + self.worker = Worker( + self.db_path, "/fake/host.py", env={"XCM_API_KEY": "x"}, + task_timeout=1800) + self.worker.executor = LocalExecutor( + "/fake/host.py", config_path=cfg_path) + _insert_task(self.db_path, "T1", goal_mode=1, goal="g", + verify_command=None, goal_max_rounds=0) + self.task = {"id": "T1", "project_id": "p1", + "_goal_resume_breakpoint": True} + + def tearDown(self): + self.tmp.cleanup() + + def _checkpoint(self, task_id, round_no): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO task_checkpoints(task_id, round_no," + " thread_id, goal_status, state_sha, verify_out," + " created_at) VALUES(?, ?, ?, ?, ?, ?, ?)", + (task_id, round_no, "th1", "goal-running", "sha", "", + 0)) + conn.commit() + finally: + conn.close() + + def test_all_absent_is_never_started(self): + self.assertTrue(self.worker._task_never_started(self.task)) + + def test_entry_present_is_not_never_started(self): + with open(self.state_path, "w", encoding="utf-8") as fh: + json.dump({"version": 1, "tasks": {"T1": {}}}, fh) + self.assertFalse(self.worker._task_never_started(self.task)) + + def test_checkpoint_present_is_not_never_started(self): + self._checkpoint("T1", 2) + self.assertFalse(self.worker._task_never_started(self.task)) + + def test_worktree_present_is_not_never_started(self): + os.makedirs(os.path.join(self.state_dir, "host-worktrees", "T1")) + self.assertFalse(self.worker._task_never_started(self.task)) + + def test_trace_present_is_not_never_started(self): + os.makedirs(os.path.join(self.state_dir, "host-traces")) + with open(os.path.join(self.state_dir, "host-traces", + "T1.jsonl"), "w", encoding="utf-8") as fh: + fh.write("{}\n") + self.assertFalse(self.worker._task_never_started(self.task)) + + def test_corrupt_state_is_not_never_started(self): + with open(self.state_path, "w", encoding="utf-8") as fh: + fh.write("{torn") + self.assertFalse(self.worker._task_never_started(self.task)) + + def test_resolve_falls_back_to_fresh_when_never_started(self): + out = self.worker._resolve_breakpoint_resume(self.task) + self.assertFalse(out.get("_goal_resume_breakpoint")) + + def test_resolve_keeps_breakpoint_when_entry_present(self): + with open(self.state_path, "w", encoding="utf-8") as fh: + json.dump({"version": 1, "tasks": {"T1": {}}}, fh) + out = self.worker._resolve_breakpoint_resume(self.task) + self.assertTrue(out.get("_goal_resume_breakpoint")) + + def test_resolve_keeps_breakpoint_when_rebuilt(self): + self._checkpoint("T1", 3) + out = self.worker._resolve_breakpoint_resume(self.task) + self.assertTrue(out.get("_goal_resume_breakpoint")) + with open(self.state_path, encoding="utf-8") as fh: + entry = json.load(fh)["tasks"]["T1"] + self.assertTrue(entry["rebuilt_from_checkpoint"]) + + def test_resolve_keeps_breakpoint_when_worktree_present(self): + os.makedirs(os.path.join(self.state_dir, "host-worktrees", "T1")) + out = self.worker._resolve_breakpoint_resume(self.task) + self.assertTrue(out.get("_goal_resume_breakpoint")) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 1bf81da..05878dc 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -1245,6 +1245,78 @@ class Worker: except Exception: return False + def _task_never_started(self, task): + """从未启动判据（2026-10-29 实测任务 A 的教训）。 + + 四者全无才算从未启动：host-state 里无本任务条目、 + 库里无断点行、state 同目录 host-worktrees 下无本任务 + worktree、host-traces 下无本任务链路文件。任一存在 + 都说明执行过，绝不能当成全新任务重跑（会撞已存在 + 或丢进度），调用方保持原有拒绝/重建路径。 + 任何一步查不出来（state 路径不明、文件损坏、库异常） + 都返回 False——fail-closed，宁可拒绝也不误重跑。 + """ + try: + state_path = self._host_state_path_for_task(task) + if not state_path: + return False + host_name = _worker_host_name(self.executor, task) + if os.path.exists(state_path): + with open(state_path, encoding="utf-8") as fh: + state = json.load(fh) + if host_name in (state.get("tasks") or {}): + return False + conn = _db_connect(self.db_path) + try: + self._ensure_result_columns(conn) + row = conn.execute( + "SELECT 1 FROM task_checkpoints WHERE task_id = ?" + " LIMIT 1", (task["id"],)).fetchone() + if row: + return False + finally: + conn.close() + state_dir = os.path.dirname(os.path.abspath(state_path)) + if os.path.isdir(os.path.join(state_dir, "host-worktrees", + host_name)): + return False + if os.path.exists(os.path.join(state_dir, "host-traces", + host_name + ".jsonl")): + return False + return True + except Exception: + return False + + def _resolve_breakpoint_resume(self, task): + """断点续跑解析：先重建，建不出且从未启动时回落全新跑。 + + 正常情况（条目在，或库里有断点可重建）原样返回， + 调用方走 goal-resume 断点续跑。只有重建失败（库里 + 也无断点）且 _task_never_started 全无执行痕迹时， + 才清掉断点标记、改走全新 goal：此时 goal-resume 注定 + 被 host 拒绝 → failed，而全新跑与首轮完全相同 + （worktree 本就不存在，不会覆盖任何进度）。 + 次数上限与无进展熔断仍在上游卡着，不会空转。 + 手动 /goal-resume 同样走这里：对从未跑过的任务， + 全新跑也是唯一有意义的结果。 + """ + try: + rebuilt = self._rebuild_host_state_from_checkpoint(task) + except Exception: + rebuilt = False + if not rebuilt and self._task_never_started(task): + task = dict(task) + task["_goal_resume_breakpoint"] = False + self._audit(task.get("project_id"), + "worker_fresh_restart", task["id"]) + try: + log_event(self.logger, "worker_fresh_restart", + task_id=task["id"], + project_id=task.get("project_id")) + except Exception: + pass + return task + def _ensure_result_columns(self, conn): """worker 侧的迁移兜底：与 server.migrate_tasks_table 同列。 @@ -4236,12 +4308,12 @@ class Worker: task["_goal_resume_breakpoint"] = True # P0 三件其一：断点丢失重建——host-state 里条目已丢 # 但库里有断点时，先按库断点拼最小条目写回，再由 - # 正常 goal-resume 续跑；库里也没有断点时返回 False， - # 正常路径继续（host 拒绝 → failed，不造假断点）。 - try: - self._rebuild_host_state_from_checkpoint(task) - except Exception: - pass + # 正常 goal-resume 续跑；库里也没有断点时返回 False。 + # 从未启动回落（2026-10-29 实测任务 A）：重建失败 + # 且四者全无执行痕迹时，goal-resume 注定被 host + # 拒绝 → failed，改走全新 goal（见 + # _resolve_breakpoint_resume），不造假断点。 + task = self._resolve_breakpoint_resume(task) try: conn = _db_connect(self.db_path) try: