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

@@ -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