Merge pull request #144 from openqj/feat-neverstart-fresh

worker: 从未启动的任务自动续跑回落全新跑

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):

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()

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: