worker: auto-resume falls back to fresh goal for never-started tasks

2026-10-29 live test: task A was interrupted 61s after claim but had
never written anything (no host-state entry, no worktree, no trace,
no checkpoint); auto-resume ran goal-resume, which host rightly
refused, and the task failed. Refusal only makes sense for tasks
with progress but a lost breakpoint (those go through the P0
checkpoint rebuild); for a never-started task the only meaningful
outcome is a fresh run, identical to its first attempt (same task
id, same goal, worktree does not exist yet so nothing is
overwritten).

Change (claim path only, via _resolve_breakpoint_resume):
- rebuild from checkpoint first, unchanged;
- if rebuild fails AND _task_never_started holds (state entry,
  checkpoint row, worktree dir, trace file all absent), clear the
  breakpoint flag so the run takes the fresh goal path, with a
  worker_fresh_restart audit event;
- any evidence of prior execution, or any check failure,
  keeps the old refusal path (fail-closed); the auto-resume
  count cap and no-progress stall guard still bound retries.

Tests: cloud/tests/test_neverstart_fresh.py, 10 items (fallback
holds, entry/checkpoint/worktree/trace presence each block it,
corrupt state blocks it, rebuild still wins when a checkpoint
exists); test_goal_resume's breakpoint-routing test now sets up
a task with a checkpoint so it still exercises goal-resume.
Full suites: cloud 1178 (skipped 4), host 104, deploy 39, green.

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