diff --git a/cloud/server.py b/cloud/server.py index ff69ba4..1e0f08e 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -1113,6 +1113,19 @@ def migrate_tasks_table(conn): # 话发一条新消息、开新循环”，本标记是“从 # goal_rounds_done 的下一轮接着跑同一循环”。 "goal_resume_pending": "INTEGER DEFAULT 0", + # 长任务自动恢复：paused_by_user=1 只由用户暂停/ + # 取消（/pause、/cancel）写，系统中断（sweep、worker + # 退出、配额除外）绝不写它；interrupted_reason 记 + # 中断来源（user_pause/user_cancel/sweep/quota/ + # coordinator_orphan/worker_shutdown），NULL = 老行或 + # 未分类系统中断，按可自动恢复处理；auto_resume_count + # 记自动续跑次数（防确定性失败无限循环的上限门）； + # no_time_limit=1 的任务 worker 级不限时（见 worker + # _run_attempt，host 单轮 600 秒仍是单轮护栏）。 + "paused_by_user": "INTEGER DEFAULT 0", + "interrupted_reason": "TEXT", + "auto_resume_count": "INTEGER DEFAULT 0", + "no_time_limit": "INTEGER DEFAULT 0", } for name, coltype in new_cols.items(): if name not in cols: @@ -1635,6 +1648,10 @@ def _task_to_dict(row, attachments=None): "parent_task_id": _row_get(row, "parent_task_id"), "coordinator_part": _row_get(row, "coordinator_part"), "goal_resume_pending": _row_get(row, "goal_resume_pending") or 0, + "paused_by_user": _row_get(row, "paused_by_user") or 0, + "interrupted_reason": _row_get(row, "interrupted_reason"), + "auto_resume_count": _row_get(row, "auto_resume_count") or 0, + "no_time_limit": _row_get(row, "no_time_limit") or 0, } @@ -1659,6 +1676,8 @@ def sweep_stale_tasks(db_path, threshold_seconds=300): 心跳取 last_heartbeat，无心跳则用 updated_at；早于 now - threshold_seconds 即过期。审计记一条 "sweep_interrupted" （user_id 记 "system"，因为回收不是某个用户的操作）。 + 这是系统中断：interrupted_reason='sweep'、paused_by_user=0， + 长任务自动恢复可接它（用户暂停绝不走这条路）。 独立可单测，不依赖 HTTP。 """ now = time.time() @@ -1674,7 +1693,9 @@ def sweep_stale_tasks(db_path, threshold_seconds=300): swept = [] for row in rows: cur = conn.execute( - "UPDATE tasks SET status = 'interrupted', updated_at = ?" + "UPDATE tasks SET status = 'interrupted'," + " interrupted_reason = 'sweep', paused_by_user = 0," + " updated_at = ?" " WHERE id = ? AND status = 'running'", (now, row["id"]), ) @@ -2059,6 +2080,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._task_agents(user_id, rest[: -len("/agents")]) elif rest.endswith("/cancel") and method == "POST": self._cancel_task(user_id, rest[: -len("/cancel")]) + elif rest.endswith("/pause") and method == "POST": + self._pause_task(user_id, rest[: -len("/pause")]) elif rest.endswith("/merge") and method == "POST": self._merge_task(user_id, rest[: -len("/merge")]) elif rest.endswith("/pr") and method == "POST": @@ -4213,6 +4236,13 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): json.dumps(coord_parts) if coord_parts else None, None, None, child_model, child_effort), ) + # 长任务不限时：API 字段 no_time_limit=1 时写标记， + # worker 认领后按不限时跑（单轮 host 600 秒仍在）； + # 不传或非真值时保持默认 0，现有任务行为不变。 + if body.get("no_time_limit") in (True, 1): + conn.execute( + "UPDATE tasks SET no_time_limit = 1" + " WHERE id = ?", (task_id,)) for rel_path, size in bundle_files: conn.execute( "INSERT INTO task_attachments" @@ -6442,8 +6472,12 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): now = time.time() conn = self._db() try: + # 用户取消 = 用户暂停：自动恢复绝不接它，人工 + # /retry 或 /goal-resume 才能再跑（那两处会清标记）。 conn.execute( - "UPDATE tasks SET status = 'interrupted', updated_at = ?" + "UPDATE tasks SET status = 'interrupted'," + " paused_by_user = 1, interrupted_reason = 'user_cancel'," + " updated_at = ?" " WHERE id = ?", (now, task_id), ) @@ -6476,6 +6510,49 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): store.audit(user_id, task["project_id"], "cancel_task", task_id) self._send_json(200, task) + def _pause_task(self, user_id, task_id): + """用户暂停：与 /cancel 同终态、不同语义标记。 + + 只有用户暂停才暂停：本端点把任务标 interrupted 并置 + paused_by_user=1、interrupted_reason='user_pause'，worker + 自动恢复（auto_resume_interrupted）绝不接它；人工 + /retry 或 /goal-resume 才能再跑（那两处会清标记）。 + 权限与 /cancel 同款（create_task），终态任务 400。 + """ + row = self._fetch_task(task_id) + if row is None: + self._send_json(404, {"error": "task not found"}) + return + task = _task_to_dict(row) + with _auth_store(self.server.db_path) as store: + if not store.check_permission( + user_id, task["project_id"], "create_task"): + self._send_json(403, {"error": "forbidden"}) + return + if task["status"] in TERMINAL_STATUSES: + self._send_json(400, {"error": "task already terminal"}) + return + now = time.time() + conn = self._db() + try: + conn.execute( + "UPDATE tasks SET status = 'interrupted'," + " paused_by_user = 1, interrupted_reason = 'user_pause'," + " updated_at = ? WHERE id = ?", + (now, task_id), + ) + conn.commit() + task = _task_to_dict( + conn.execute( + "SELECT * FROM tasks WHERE id = ?", (task_id,) + ).fetchone() + ) + finally: + conn.close() + with _auth_store(self.server.db_path) as store: + store.audit(user_id, task["project_id"], "pause_task", task_id) + self._send_json(200, task) + def _merge_task(self, user_id, task_id): """把云端任务标记为已合并：只补状态，不执行合并本身。 @@ -6924,7 +7001,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): cur = conn.execute( "UPDATE tasks SET status = 'queued'," " goal_resume_pending = 1, result_text = NULL," - " error_text = NULL, actual_model = NULL, updated_at = ?" + " error_text = NULL, actual_model = NULL," + " paused_by_user = 0, interrupted_reason = NULL," + " updated_at = ?" " WHERE id = ? AND status = 'interrupted'", (now, task_id), ) @@ -6993,7 +7072,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): try: cur = conn.execute( "UPDATE tasks SET status = 'queued', result_text = NULL," - " error_text = NULL, actual_model = NULL, updated_at = ?" + " error_text = NULL, actual_model = NULL," + " paused_by_user = 0, interrupted_reason = NULL," + " updated_at = ?" " WHERE id = ? AND status IN" " ('failed', 'interrupted', 'merge-failed')", (now, task_id), diff --git a/cloud/tests/test_longtask_auto_resume.py b/cloud/tests/test_longtask_auto_resume.py new file mode 100644 index 0000000..c90c426 --- /dev/null +++ b/cloud/tests/test_longtask_auto_resume.py @@ -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() diff --git a/cloud/worker.py b/cloud/worker.py index d4486b6..50ad07b 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -360,6 +360,12 @@ def _task_from_row(row): "resume_message": _col("resume_message"), # G10：goal 断点续跑标记，老行缺列时 0 = 普通排队。 "goal_resume_pending": _col("goal_resume_pending") or 0, + # 长任务自动恢复：用户暂停标记、中断原因、自动续跑 + # 次数、不限时标记，老行缺列时按默认（可恢复、不限时关）。 + "paused_by_user": _col("paused_by_user") or 0, + "interrupted_reason": _col("interrupted_reason"), + "auto_resume_count": _col("auto_resume_count") or 0, + "no_time_limit": _col("no_time_limit") or 0, # 协调员：老行缺列时 coordinator_parts 为 []（= 普通任务）， # parent_task_id/coordinator_part 缺列时为 None。 "coordinator_parts": _coord_parse_parts(_col("coordinator_parts")), @@ -395,6 +401,7 @@ class Worker: outbox_backoff_seconds=60, outbox_max_attempts=3, quota_poll_interval_seconds=30, + auto_resume_max=50, ignore_host_budget=True): self.db_path = str(db_path) self.host_py_path = host_py_path @@ -494,6 +501,11 @@ class Worker: self.task_timeout = task_timeout self.max_retries = max_retries self.retry_backoff_seconds = retry_backoff_seconds + # 长任务自动恢复上限（次数，不是时间）：同一任务被 + # 系统中断后自动续跑超过该次数即停住、等人工处理， + # 防确定性失败（每次续跑立刻又中断）无限烧额度； + # 正常部署重启/偶发中断远达不到 50 次。 + self.auto_resume_max = auto_resume_max # 项目仓库检出根目录：CLI --repos-dir 可调（默认 ./repos）； # Docker 执行器下它必须放在 extra-mounts 已挂载的目录下， # 否则容器内 task["repo"] 路径不存在——见 cloud/README.md。 @@ -813,6 +825,96 @@ class Worker: finally: conn.close() + def _mark_interrupted_reason(self, task_id, reason): + """给已是 interrupted 的任务补中断原因（不改状态本身）。 + + 只写 reason，不碰 paused_by_user：调用方是系统路径 + （配额、sweep 已在 server 侧写、worker 退出、协调员 + 孤儿），用户暂停的 paused 标记由 server /pause、 + /cancel 写，worker 侧绝不覆盖它。 + 老库缺列时先迁移，迁移失败则静默跳过（不炸终局路径）。 + """ + try: + conn = _db_connect(self.db_path) + try: + self._ensure_result_columns(conn) + conn.execute( + "UPDATE tasks SET interrupted_reason = ?" + " WHERE id = ? AND status = 'interrupted'", + (reason, task_id), + ) + conn.commit() + finally: + conn.close() + except Exception: + pass + + def auto_resume_interrupted(self): + """长任务自动恢复：只接系统中断的 goal 任务。 + + 只有用户暂停才暂停：paused_by_user=1（/pause、/cancel） + 的任务绝不自动动；配额中断（quota）与协调员孤儿 + （coordinator_orphan）也不接——前者要等配额恢复、 + 后者要由父协调员收口，自动重跑只会烧钱或跑丢父子关系。 + 非 goal 任务不接：它没有断点语义，重跑等于从头再来， + 可能重复副作用，留给人工 /retry 决定。 + 接到的任务 CAS 回 queued 并置 goal_resume_pending=1， + 下一轮认领即走 G10 断点续跑（同一线程、同一 worktree、 + 从 goal_rounds_done 下一轮起），同时清旧结果、清 + interrupted_reason、auto_resume_count+1；上限 + self.auto_resume_max 到顶的任务停住并记审计，等人工。 + 返回自动续跑的任务 id 列表；任何异常只记日志，不炸 + run_loop（恢复是辅助路径，不能带走常驻 worker）。 + """ + resumed = [] + try: + conn = _db_connect(self.db_path) + try: + self._ensure_result_columns(conn) + rows = conn.execute( + "SELECT id, project_id 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" + " ('quota', 'coordinator_orphan'))" + " AND parent_task_id IS NULL" + " AND (goal_mode = 1 OR verify_command IS NOT NULL" + " OR goal IS NOT NULL)" + " AND (auto_resume_count IS NULL OR" + " auto_resume_count < ?)", + (self.auto_resume_max,), + ).fetchall() + now = time.time() + for row in rows: + task_id = row[0] + project_id = row[1] + cur = conn.execute( + "UPDATE tasks SET status = 'queued'," + " goal_resume_pending = 1," + " result_text = NULL, error_text = NULL," + " actual_model = NULL," + " interrupted_reason = NULL," + " auto_resume_count =" + " COALESCE(auto_resume_count, 0) + 1," + " updated_at = ?" + " WHERE id = ? AND status = 'interrupted'" + " AND (paused_by_user IS NULL" + " OR paused_by_user = 0)", + (now, task_id), + ) + if cur.rowcount == 1: + resumed.append(task_id) + self._audit(project_id, "worker_auto_resumed", + task_id) + conn.commit() + finally: + conn.close() + except Exception as exc: + log_event(self.logger, "worker_auto_resume_failed", + message="auto resume failed: %s" % exc) + return resumed + def _ensure_result_columns(self, conn): """worker 侧的迁移兜底：与 server.migrate_tasks_table 同列。 @@ -851,7 +953,11 @@ class Worker: "coordinator_parts": "TEXT", "parent_task_id": "TEXT", "coordinator_part": "TEXT", - "goal_resume_pending": "INTEGER DEFAULT 0"}.items(): + "goal_resume_pending": "INTEGER DEFAULT 0", + "paused_by_user": "INTEGER DEFAULT 0", + "interrupted_reason": "TEXT", + "auto_resume_count": "INTEGER DEFAULT 0", + "no_time_limit": "INTEGER DEFAULT 0"}.items(): if name not in cols: try: conn.execute( @@ -2930,21 +3036,29 @@ class Worker: container=self.executor.container_name(task)) except Exception: pass + # 长任务不限时：任务带 no_time_limit=1 时，本次 worker + # 级与 Docker 级超时一律按 None 算（不限时），不改全局 + # task_timeout（其它任务仍走默认 1800 秒）；host 单轮 + # turn_timeout（默认 600 秒）不在 worker 侧传参，仍是 + # 单轮护栏——不限的是任务总墙钟与总轮次，不是单轮无限。 + _no_time_limit = bool(task.get("no_time_limit")) + _effective_task_timeout = None if _no_time_limit \ + else self.task_timeout # Docker 超时取较小值：两个超时都不应静默延长另一个。 # 注意绝不能改共享的 executor.timeout（并发下两个任务会 # 互相覆盖对方的超时）；超时只作为本次调用的参数传下去。 effective_docker_timeout = None is_docker = isinstance(self.executor, DockerExecutor) - if is_docker and self.task_timeout is not None: + if is_docker and _effective_task_timeout is not None: if self.executor.timeout is None: - effective_docker_timeout = self.task_timeout + effective_docker_timeout = _effective_task_timeout else: try: effective_docker_timeout = min( - float(self.task_timeout), + float(_effective_task_timeout), float(self.executor.timeout)) except (TypeError, ValueError): - effective_docker_timeout = self.task_timeout + effective_docker_timeout = _effective_task_timeout # P0 安全批后：Docker 同步路径起取消看门狗线程 # （阻塞在 docker run 里时，库状态变 interrupted 就 # 按容器名真杀；非 Docker 路径不启动，避免空线程）。 @@ -2988,7 +3102,7 @@ class Worker: popen_kind = "dispatch" result = self._execute_via_process( task, kind=popen_kind, - deadline_seconds=self.task_timeout, + deadline_seconds=_effective_task_timeout, seed_host_dir=seed_host_dir, quota_check=quota_check) combined = (result.get("stdout", "") or "") + \ @@ -2999,7 +3113,7 @@ class Worker: _ALREADY_EXISTS_MARKER in combined: result = self._execute_via_process( task, kind="run", - deadline_seconds=self.task_timeout, + deadline_seconds=_effective_task_timeout, quota_check=quota_check) else: if is_docker: @@ -3428,6 +3542,7 @@ class Worker: for ch in pending: cur = conn.execute( "UPDATE tasks SET status = 'interrupted'," + " interrupted_reason = 'coordinator_orphan'," " updated_at = ? WHERE id = ? AND status NOT IN" " ('completed', 'failed', 'interrupted', 'merged'," " 'merge-failed')", @@ -3490,6 +3605,7 @@ class Worker: continue cur = conn.execute( "UPDATE tasks SET status = 'interrupted'," + " interrupted_reason = 'coordinator_orphan'," " updated_at = ? WHERE id = ? AND status NOT IN" " ('completed', 'failed', 'interrupted', 'merged'," " 'merge-failed')", @@ -3830,6 +3946,9 @@ class Worker: quota_error = self._quota_exceeded(task, None, on_error="closed") if quota_error is not None: self._set_status(task_id, "interrupted") + # 配额中断不自动恢复：要等配额恢复/调高，自动 + # 回 queued 只会让 autoscaler 按队列深度白拉 worker。 + self._mark_interrupted_reason(task_id, "quota") self._write_execution_result( task_id, error_text=quota_error, actual_model=self.model, started_at=started_at, finished_at=time.time(), @@ -4280,6 +4399,8 @@ class Worker: self._audit(task["project_id"], "worker_fallback", "%s -> %s" % (self.model, nxt)) final_written = self._set_status_if_running(task_id, final) + if quota_stop and final_written and final == "interrupted": + self._mark_interrupted_reason(task_id, "quota") finished_at = time.time() # P0 安全批后：CAS 没改到行（任务已被取消/回收）时 # 绝不写回结果，否则迟到的旧进程会覆盖新一轮的结果 @@ -4764,6 +4885,13 @@ class Worker: except Exception as exc: log_event(self.logger, "worker_orphan_sweep_failed", message="orphan sweep failed: %s" % exc) + # 长任务自动恢复：接在 sweep 之后——刚被 sweep + # 标 interrupted 的系统中断 goal 任务，本轮即可 + # 回 queued、下一轮认领即断点续跑；用户暂停、 + # 配额、协调员孤儿一律不接（见方法 docstring）。 + # 放在 sweep 块外：sweep 关掉时，已由其它路径 + # （如 worker 退出）标 interrupted 的任务仍要恢复。 + self.auto_resume_interrupted() # 每轮结束写心跳：哪怕本轮一个任务都没派，worker 活着 # 这件事本身也要被 server 看到（文件心跳 + 注册表心跳）。 self._write_heartbeat() @@ -4810,6 +4938,8 @@ class Worker: pass self._running_clear(task_id) self._set_status(task_id, "interrupted") + # worker 退出时的系统中断，可自动恢复。 + self._mark_interrupted_reason(task_id, "worker_shutdown") return all_results @@ -4970,6 +5100,10 @@ def main(argv=None): "取消与已记 fallback 的失败不重试）") parser.add_argument("--retry-backoff-seconds", type=float, default=5.0, help="重试前等待秒数（默认 5）") + parser.add_argument("--auto-resume-max", type=int, default=50, + help="长任务自动恢复上限（次数，非时间，默认 50）：" + "同一 goal 任务被系统中断后自动续跑超过该次数" + "即停住等人工，防确定性失败无限烧额度") parser.add_argument("--loop", action="store_true", help="常驻循环，每 --interval 秒跑一轮") parser.add_argument("--interval", type=int, default=300) @@ -5002,6 +5136,7 @@ def main(argv=None): task_timeout=args.task_timeout, max_retries=args.max_retries, retry_backoff_seconds=args.retry_backoff_seconds, + auto_resume_max=args.auto_resume_max, repos_dir=args.repos_dir, heartbeat_file=args.heartbeat_file, worker_id=args.worker_id,