Merge pull request #201 from openqj/feat-platform-auto-resume

feat: auto-resume platform-interrupted non-goal tasks

cloud/tests/test_longtask_auto_resume.py

@@ -5,7 +5,9 @@
 - 系统中断的 goal 任务 -> 回 queued + goal_resume_pending=1 + count+1
 - 用户暂停/取消（paused_by_user=1）-> 不动
 - 配额仍超限 -> 不动（不重排队）；配额已恢复 -> 自动续跑
-- 协调员孤儿、非 goal 任务 -> 不动
+- 协调员孤儿、非平台中断的非 goal 任务 -> 不动
+- 平台中断（worker_shutdown/sweep）的非 goal 任务 -> 回 queued
+  全新跑（goal_resume_pending=0，resume_message 清掉）
 - 自动续跑次数到上限 -> 不动
 - no_time_limit 任务的 _run_attempt 不设 worker 级截止
 """
@@ -64,6 +66,17 @@ def _row(db_path, task_id):
         conn.close()
 
 
+def _resume_msg(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        row = conn.execute(
+            "SELECT resume_message FROM tasks WHERE id = ?",
+            (task_id,)).fetchone()
+        return row[0] if row else None
+    finally:
+        conn.close()
+
+
 class LongtaskAutoResumeTestCase(unittest.TestCase):
     def setUp(self):
         self.tmp = tempfile.TemporaryDirectory()
@@ -115,6 +128,38 @@ class LongtaskAutoResumeTestCase(unittest.TestCase):
             status, _pending, _c, _p, _r, _n = _row(self.db_path, tid)
             self.assertEqual(status, "interrupted")
 
+    def test_platform_interrupted_nongoal_resumes_fresh(self):
+        # 平台中断的非 goal 任务：全新跑，不置断点标记，
+        # 残留的续跑消息清掉，旧断点标记清掉。
+        _insert_task(self.db_path, "T1", goal_mode=0,
+                     interrupted_reason="worker_shutdown",
+                     paused_by_user=0, goal_resume_pending=1,
+                     resume_message="stale-msg")
+        _insert_task(self.db_path, "T2", goal_mode=0,
+                     interrupted_reason="sweep", paused_by_user=0)
+        resumed = self.worker.auto_resume_interrupted()
+        self.assertEqual(sorted(resumed), ["T1", "T2"])
+        for tid in ("T1", "T2"):
+            status, pending, count, _p, reason, _n = _row(
+                self.db_path, tid)
+            self.assertEqual(status, "queued")
+            self.assertEqual(pending, 0)
+            self.assertEqual(count, 1)
+            self.assertIsNone(reason)
+        self.assertIsNone(_resume_msg(self.db_path, "T1"))
+
+    def test_other_reason_nongoal_does_not_resume(self):
+        # 非平台中断的非 goal 任务仍不接：原因 NULL、配额
+        # （即使配额已恢复，配额口径只放 goal 任务）。
+        self.worker._quota_exceeded = lambda task, usage, on_error="closed": None
+        _insert_task(self.db_path, "T1", goal_mode=0)
+        _insert_task(self.db_path, "T2", goal_mode=0,
+                     interrupted_reason="quota", paused_by_user=0)
+        self.assertEqual(self.worker.auto_resume_interrupted(), [])
+        for tid in ("T1", "T2"):
+            status, _pending, _c, _p, _r, _n = _row(self.db_path, tid)
+            self.assertEqual(status, "interrupted")
+
     def test_quota_restored_resumes(self):
         # 配额已恢复（现查为 None）时配额中断也自动续跑。
         self.worker._quota_exceeded = lambda task, usage, on_error="closed": None

cloud/worker.py

@@ -923,7 +923,8 @@ class Worker:
             pass
 
     def auto_resume_interrupted(self):
-        """长任务自动恢复：只接系统中断的 goal 任务。
+        """长任务自动恢复：接系统中断的 goal 任务，以及平台
+        中断的非 goal 任务。
 
         只有用户暂停才暂停：paused_by_user=1（/pause、/cancel）
         的任务绝不自动动；协调员孤儿（coordinator_orphan）
@@ -934,8 +935,18 @@ class Worker:
         下一轮再查；查不到配额或仍超限都不接。恢复的判定
         是现查结果为 None（新月份、额度上调、无价上限重置
         都算恢复），不是猜时间。
-        非 goal 任务不接：它没有断点语义，重跑等于从头再来，
-        可能重复副作用，留给人工 /retry 决定。
+        非 goal 任务只接平台中断：interrupted_reason 为
+        worker_shutdown（worker 关停，如部署重启）或 sweep
+        （server 侧陈旧回收，如 worker 被强杀）时，平台杀了
+        任务、平台负责重排；重排是全新跑（不置
+        goal_resume_pending、不走断点，同一 worktree 里文件
+        还在），并清掉可能残留的 resume_message，防旧续跑
+        消息劫持新跑的 prompt。其它原因（NULL、用户取消等）
+        的非 goal 任务仍不接，留给人工 /retry 决定。
+        注意：非 goal 任务没有断点，断点轮次恒为 -1，因此
+        每次平台中断重排都会计一次无进展——3 次重排后第 4
+        次平台中断即 stall 停住等人工，这是防 crash-重排死
+        循环的同一道闸，不是 bug。
         接到的任务 CAS 回 queued 并置 goal_resume_pending=1，
         下一轮认领即走 G10 断点续跑（同一线程、同一 worktree、
         从 goal_rounds_done 下一轮起），同时清旧结果、清
@@ -962,15 +973,18 @@ class Worker:
                     "SELECT id, project_id, created_by, parent_task_id,"
                     " model, interrupted_reason, resume_after,"
                     " auto_resume_stall_count,"
-                    " checkpoint_round_at_last_resume FROM tasks"
+                    " checkpoint_round_at_last_resume,"
+                    " goal_mode, verify_command, goal FROM tasks"
                     " WHERE status = 'interrupted'"
                     " AND (paused_by_user IS NULL OR paused_by_user = 0)"
                     " AND (interrupted_reason IS NULL OR"
                     " interrupted_reason NOT IN"
                     " ('coordinator_orphan'))"
                     " AND parent_task_id IS NULL"
-                    " AND (goal_mode = 1 OR verify_command IS NOT NULL"
+                    " AND ((goal_mode = 1 OR verify_command IS NOT NULL"
                     " OR goal IS NOT NULL)"
+                    " OR interrupted_reason IN"
+                    " ('worker_shutdown', 'sweep'))"
                     " AND (auto_resume_count IS NULL OR"
                     " auto_resume_count < ?)",
                     (self.auto_resume_max,),
@@ -1002,6 +1016,8 @@ class Worker:
                     except Exception:
                         cp_round = -1
                     prev = row[8]
+                    is_goal = (row[9] == 1 or row[10] is not None
+                               or row[11] is not None)
                     stall = 0
                     if prev is not None:
                         stall = (row[7] or 0) + 1 if cp_round == prev else 0
@@ -1019,10 +1035,17 @@ class Worker:
                                 task_id)
                         continue
                     backoff = min(300, 10 * (2 ** stall))
+                    if is_goal:
+                        resume_set = "goal_resume_pending = 1"
+                    else:
+                        # 全新跑：不置断点标记，并清掉残留的续跑
+                        # 消息，防它劫持新跑的 prompt。
+                        resume_set = ("goal_resume_pending = 0,"
+                                      " resume_message = NULL")
                     cur = conn.execute(
-                        "UPDATE tasks SET status = 'queued',"
-                        " goal_resume_pending = 1,"
-                        " result_text = NULL, error_text = NULL,"
+                        "UPDATE tasks SET status = 'queued', "
+                        + resume_set +
+                        ", result_text = NULL, error_text = NULL,"
                         " actual_model = NULL,"
                         " interrupted_reason = NULL,"
                         " auto_resume_count ="
@@ -1040,10 +1063,16 @@ class Worker:
                         resumed.append(task_id)
                         self._audit(project_id, "worker_auto_resumed",
                                     task_id)
+                        if is_goal:
+                            summary = ("自动续跑（无进展计数 %d，"
+                                       "下次退避 %d 秒）"
+                                       % (stall, backoff))
+                        else:
+                            summary = ("平台中断自动重排（%s，全新跑，"
+                                       "无进展计数 %d，下次退避 %d 秒）"
+                                       % (row[5], stall, backoff))
                         self._record_step(
-                            conn, task_id, "auto_resume",
-                            "自动续跑（无进展计数 %d，下次退避 %d 秒）"
-                            % (stall, backoff))
+                            conn, task_id, "auto_resume", summary)
                 conn.commit()
             finally:
                 conn.close()