From 4428a721a6f6921351463f3baf0f773d2b617f6f Mon Sep 17 00:00:00 2001 From: host Date: Fri, 30 Oct 2026 15:39:55 +0800 Subject: [PATCH] feat: auto-resume platform-interrupted non-goal tasks Deploy restarts kill running tasks: the worker marks them interrupted (worker_shutdown on graceful exit, sweep on stale reclaim), and auto_resume_interrupted used to accept goal tasks only, so every non-goal task killed by a deploy stayed interrupted forever unless a human retried it (the 2026-10-30 demo task died this way). Now worker_shutdown/sweep interruptions are re-queued for non-goal tasks too: the platform killed the task, so the platform re-queues it. The re-queue is a fresh run, not a breakpoint resume: goal_resume_pending is forced to 0 and any stale resume_message is cleared, so the new run cannot be hijacked by the goal-resume path or an old continue message; files in the same worktree persist. All existing guards stay: user pause, coordinator orphans, quota (still goal-only), backoff, the 3-strike stall guard (non-goal tasks have no checkpoints, so each platform re-queue counts as no-progress), and the auto_resume_max cap. Other non-goal reasons (NULL, quota, user cancel) still do not auto-resume. --- cloud/tests/test_longtask_auto_resume.py | 47 +++++++++++++++++++++- cloud/worker.py | 51 +++++++++++++++++++----- 2 files changed, 86 insertions(+), 12 deletions(-) diff --git a/cloud/tests/test_longtask_auto_resume.py b/cloud/tests/test_longtask_auto_resume.py index bc443bd..f38361d 100644 --- a/cloud/tests/test_longtask_auto_resume.py +++ b/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 diff --git a/cloud/worker.py b/cloud/worker.py index cdb35dd..64a9824 100644 --- a/cloud/worker.py +++ b/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()