feat: 长任务中断自动恢复，仅用户暂停才暂停，不限时
- 新增 tasks 列 paused_by_user/interrupted_reason/auto_resume_count/no_time_limit（幂等迁移，server+worker 双侧）
- POST /tasks/{id}/pause：用户暂停标 interrupted + paused_by_user=1；/cancel 同款标记；人工 /retry、/goal-resume 清标记
- sweep 标 sweep、配额标 quota、协调员孤儿标 coordinator_orphan、worker 退出标 worker_shutdown
- worker.run_loop 每轮后 auto_resume_interrupted：只接系统中断的 goal 任务，CAS 回 queued + goal_resume_pending=1（G10 断点续跑），用户暂停/配额/孤儿/非 goal 不接，上限 50 次防无限烧额度
- no_time_limit=1 的任务 worker/Docker 级不限时（其它任务仍默认 1800 秒；host 单轮 600 秒仍是单轮护栏）
- 前置 cherry-pick：G10 断点续跑透传 model_provider（否则 Docker 断点续跑在 sub2api 下必死，自动恢复无法实证；本 PR supersedes #127，勿另行合并 #127）
This commit is contained in:
host committed
1 parent
a19b1567bb
commit
c9cfe22a36
3 files changed
+454
-11
No files matched your search
@@ -0,0 +1,227 @@
|
||||
"""长任务自动恢复单测：只有用户暂停才暂停。
|
||||
|
||||
只测本路改动，不跑真实模型回合、不联网。
|
||||
规则（见 worker.auto_resume_interrupted docstring）：
|
||||
- 系统中断的 goal 任务 -> 回 queued + goal_resume_pending=1 + count+1
|
||||
- 用户暂停/取消（paused_by_user=1）-> 不动
|
||||
- 配额中断、协调员孤儿、非 goal 任务 -> 不动
|
||||
- 自动续跑次数到上限 -> 不动
|
||||
- no_time_limit 任务的 _run_attempt 不设 worker 级截止
|
||||
"""
|
||||
|
||||
import http.client
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
from cloud.auth import AuthStore
|
||||
from cloud.server import create_server, init_db
|
||||
from cloud.worker import Worker
|
||||
except ImportError:
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||||
from cloud.auth import AuthStore
|
||||
from cloud.server import create_server, init_db
|
||||
from cloud.worker import Worker
|
||||
|
||||
|
||||
def _insert_task(db_path, task_id, status="interrupted", **cols):
|
||||
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", status, "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()
|
||||
|
||||
|
||||
def _row(db_path, task_id):
|
||||
conn = sqlite3.connect(db_path)
|
||||
try:
|
||||
row = conn.execute(
|
||||
"SELECT status, goal_resume_pending, auto_resume_count,"
|
||||
" paused_by_user, interrupted_reason, no_time_limit"
|
||||
" FROM tasks WHERE id = ?", (task_id,)).fetchone()
|
||||
return row
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
class LongtaskAutoResumeTestCase(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.worker = Worker(
|
||||
self.db_path, "/fake/host.py", env={"XCM_API_KEY": "x"},
|
||||
task_timeout=1800)
|
||||
|
||||
def tearDown(self):
|
||||
self.tmp.cleanup()
|
||||
|
||||
def test_system_interrupted_goal_resumes(self):
|
||||
_insert_task(self.db_path, "T1", goal_mode=1, goal="g",
|
||||
interrupted_reason="sweep", paused_by_user=0)
|
||||
resumed = self.worker.auto_resume_interrupted()
|
||||
self.assertEqual(resumed, ["T1"])
|
||||
status, pending, count, _paused, reason, _ntl = _row(
|
||||
self.db_path, "T1")
|
||||
self.assertEqual(status, "queued")
|
||||
self.assertEqual(pending, 1)
|
||||
self.assertEqual(count, 1)
|
||||
self.assertIsNone(reason)
|
||||
|
||||
def test_user_pause_and_cancel_do_not_resume(self):
|
||||
_insert_task(self.db_path, "T1", goal_mode=1, goal="g",
|
||||
interrupted_reason="user_pause", paused_by_user=1)
|
||||
_insert_task(self.db_path, "T2", goal_mode=1, goal="g",
|
||||
interrupted_reason="user_cancel", paused_by_user=1)
|
||||
self.assertEqual(self.worker.auto_resume_interrupted(), [])
|
||||
for tid in ("T1", "T2"):
|
||||
status, pending, count, _p, _r, _n = _row(self.db_path, tid)
|
||||
self.assertEqual(status, "interrupted")
|
||||
self.assertEqual(pending, 0)
|
||||
self.assertEqual(count, 0)
|
||||
|
||||
def test_quota_orphan_nongoal_do_not_resume(self):
|
||||
_insert_task(self.db_path, "T1", goal_mode=1, goal="g",
|
||||
interrupted_reason="quota", paused_by_user=0)
|
||||
_insert_task(self.db_path, "T2", goal_mode=1, goal="g",
|
||||
interrupted_reason="coordinator_orphan",
|
||||
paused_by_user=0, parent_task_id="P0")
|
||||
_insert_task(self.db_path, "T3", goal_mode=0)
|
||||
self.assertEqual(self.worker.auto_resume_interrupted(), [])
|
||||
for tid in ("T1", "T2", "T3"):
|
||||
status, _pending, _c, _p, _r, _n = _row(self.db_path, tid)
|
||||
self.assertEqual(status, "interrupted")
|
||||
|
||||
def test_auto_resume_max_cap(self):
|
||||
_insert_task(self.db_path, "T1", goal_mode=1, goal="g",
|
||||
interrupted_reason="sweep", paused_by_user=0,
|
||||
auto_resume_count=50)
|
||||
self.assertEqual(self.worker.auto_resume_interrupted(), [])
|
||||
status, _p, count, _pa, _r, _n = _row(self.db_path, "T1")
|
||||
self.assertEqual(status, "interrupted")
|
||||
self.assertEqual(count, 50)
|
||||
|
||||
def test_no_time_limit_run_attempt_has_no_deadline(self):
|
||||
captured = {}
|
||||
|
||||
def fake_execute(task, kind="dispatch", deadline_seconds=None,
|
||||
seed_host_dir=None, quota_check=None):
|
||||
captured["deadline"] = deadline_seconds
|
||||
return {"status": "completed", "stdout": "", "stderr": ""}
|
||||
|
||||
self.worker._execute_via_process = fake_execute
|
||||
self.worker._ensure_gateway = lambda task: (None, {})
|
||||
self.worker.process_factory = object()
|
||||
task = {"id": "T1", "project_id": "p1", "prompt": "x",
|
||||
"goal_mode": 0, "no_time_limit": 1}
|
||||
self.worker._run_attempt(task)
|
||||
self.assertIsNone(captured["deadline"])
|
||||
task2 = dict(task, no_time_limit=0)
|
||||
self.worker._run_attempt(task2)
|
||||
self.assertEqual(captured["deadline"], 1800)
|
||||
|
||||
|
||||
class ServerPauseTestCase(unittest.TestCase):
|
||||
"""POST /tasks/{id}/pause：用户暂停标记 + 自动恢复不接它。"""
|
||||
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||||
self.auth = AuthStore(self.db_path)
|
||||
self.owner_id, self.owner_key = self.auth.create_user("owner")
|
||||
self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS")
|
||||
os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id
|
||||
self.server = create_server(self.db_path, "127.0.0.1", 0)
|
||||
self.port = self.server.server_address[1]
|
||||
self.thread = threading.Thread(
|
||||
target=self.server.serve_forever, daemon=True)
|
||||
self.thread.start()
|
||||
status, project = self._req(
|
||||
"POST", "/projects", {"name": "demo"}, self.owner_key)
|
||||
self.assertEqual(status, 201, project)
|
||||
self.project_id = project["id"]
|
||||
_insert_task(self.db_path, "TP", status="running",
|
||||
goal_mode=1, goal="g")
|
||||
|
||||
def tearDown(self):
|
||||
self.server.shutdown()
|
||||
self.server.server_close()
|
||||
self.thread.join(timeout=5)
|
||||
try:
|
||||
self.auth.conn.close()
|
||||
except Exception:
|
||||
pass
|
||||
self.tmp.cleanup()
|
||||
if getattr(self, "_old_admins", None) is None:
|
||||
os.environ.pop("CODEX_GLOBAL_ADMINS", None)
|
||||
else:
|
||||
os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins
|
||||
|
||||
def _req(self, method, path, body=None, key=None):
|
||||
conn = http.client.HTTPConnection(
|
||||
"127.0.0.1", self.port, timeout=10)
|
||||
hdrs = {}
|
||||
payload = None
|
||||
if body is not None:
|
||||
payload = json.dumps(body).encode("utf-8")
|
||||
hdrs["Content-Type"] = "application/json"
|
||||
if key is not None:
|
||||
hdrs["X-API-Key"] = key
|
||||
try:
|
||||
conn.request(method, path, body=payload, headers=hdrs)
|
||||
resp = conn.getresponse()
|
||||
data = resp.read()
|
||||
finally:
|
||||
conn.close()
|
||||
try:
|
||||
parsed = json.loads(data.decode("utf-8")) if data else None
|
||||
except ValueError:
|
||||
parsed = None
|
||||
return resp.status, parsed
|
||||
|
||||
def test_pause_marks_user_paused_and_auto_resume_skips(self):
|
||||
# 注意：_insert_task 的 project_id 是 "p1"，与本用例项目
|
||||
# 不同——pause 的权限校验按任务所属项目查，owner 不是
|
||||
# p1 成员会 403；所以先把任务项目改成本用例项目。
|
||||
conn = sqlite3.connect(self.db_path)
|
||||
try:
|
||||
conn.execute(
|
||||
"UPDATE tasks SET project_id = ? WHERE id = ?",
|
||||
(self.project_id, "TP"))
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
status, body = self._req(
|
||||
"POST", "/tasks/TP/pause", {}, self.owner_key)
|
||||
self.assertEqual(status, 200, body)
|
||||
self.assertEqual(body["status"], "interrupted")
|
||||
self.assertEqual(body["paused_by_user"], 1)
|
||||
self.assertEqual(body["interrupted_reason"], "user_pause")
|
||||
worker = Worker(self.db_path, "/fake/host.py",
|
||||
env={"XCM_API_KEY": "x"})
|
||||
self.assertEqual(worker.auto_resume_interrupted(), [])
|
||||
row = _row(self.db_path, "TP")
|
||||
self.assertEqual(row[0], "interrupted")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user