diff --git a/cloud/tests/test_longtask_auto_resume.py b/cloud/tests/test_longtask_auto_resume.py index c90c426..bc443bd 100644 --- a/cloud/tests/test_longtask_auto_resume.py +++ b/cloud/tests/test_longtask_auto_resume.py @@ -4,7 +4,8 @@ 规则（见 worker.auto_resume_interrupted docstring）： - 系统中断的 goal 任务 -> 回 queued + goal_resume_pending=1 + count+1 - 用户暂停/取消（paused_by_user=1）-> 不动 -- 配额中断、协调员孤儿、非 goal 任务 -> 不动 +- 配额仍超限 -> 不动（不重排队）；配额已恢复 -> 自动续跑 +- 协调员孤儿、非 goal 任务 -> 不动 - 自动续跑次数到上限 -> 不动 - no_time_limit 任务的 _run_attempt 不设 worker 级截止 """ @@ -100,6 +101,9 @@ class LongtaskAutoResumeTestCase(unittest.TestCase): self.assertEqual(count, 0) def test_quota_orphan_nongoal_do_not_resume(self): + # 配额仍超限时不重排队（mock 现查超限）。 + self.worker._quota_exceeded = lambda task, usage, on_error="closed": ( + "quota_exceeded: 本月配额已用完") _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", @@ -111,6 +115,18 @@ class LongtaskAutoResumeTestCase(unittest.TestCase): 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 + _insert_task(self.db_path, "T1", goal_mode=1, goal="g", + interrupted_reason="quota", paused_by_user=0) + self.assertEqual(self.worker.auto_resume_interrupted(), ["T1"]) + status, pending, count, _p, reason, _n = _row(self.db_path, "T1") + self.assertEqual(status, "queued") + self.assertEqual(pending, 1) + self.assertEqual(count, 1) + self.assertIsNone(reason) + 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, diff --git a/cloud/worker.py b/cloud/worker.py index 50ad07b..5be6fb9 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -853,9 +853,14 @@ class Worker: """长任务自动恢复：只接系统中断的 goal 任务。 只有用户暂停才暂停：paused_by_user=1（/pause、/cancel） - 的任务绝不自动动；配额中断（quota）与协调员孤儿 - （coordinator_orphan）也不接——前者要等配额恢复、 - 后者要由父协调员收口，自动重跑只会烧钱或跑丢父子关系。 + 的任务绝不自动动；协调员孤儿（coordinator_orphan） + 不接——要由父协调员收口，自动重跑会跑丢父子关系。 + 配额中断（quota）只在配额已恢复时接：每轮先跑 + _quota_exceeded 现查，仍超限就留在 interrupted 不 + 重排队（不触发 autoscaler 拉 worker、不烧钱），等 + 下一轮再查；查不到配额或仍超限都不接。恢复的判定 + 是现查结果为 None（新月份、额度上调、无价上限重置 + 都算恢复），不是猜时间。 非 goal 任务不接：它没有断点语义，重跑等于从头再来， 可能重复副作用，留给人工 /retry 决定。 接到的任务 CAS 回 queued 并置 goal_resume_pending=1， @@ -872,12 +877,13 @@ class Worker: try: self._ensure_result_columns(conn) rows = conn.execute( - "SELECT id, project_id FROM tasks" + "SELECT id, project_id, created_by, parent_task_id," + " model, interrupted_reason 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'))" + " ('coordinator_orphan'))" " AND parent_task_id IS NULL" " AND (goal_mode = 1 OR verify_command IS NOT NULL" " OR goal IS NOT NULL)" @@ -889,6 +895,19 @@ class Worker: for row in rows: task_id = row[0] project_id = row[1] + if row[5] == "quota": + # 配额中断：现查配额，仍超限或查不到 + # 就留在 interrupted，下轮再查，不重排队。 + try: + _qerr = self._quota_exceeded( + {"created_by": row[2], + "parent_task_id": row[3], + "model": row[4]}, + None, on_error="closed") + except Exception: + _qerr = "quota_check_failed" + if _qerr is not None: + continue cur = conn.execute( "UPDATE tasks SET status = 'queued'," " goal_resume_pending = 1," @@ -4940,6 +4959,25 @@ class Worker: self._set_status(task_id, "interrupted") # worker 退出时的系统中断，可自动恢复。 self._mark_interrupted_reason(task_id, "worker_shutdown") + # 审计：此前这条路径只改状态不记审计，中断只能靠 + # 重启时间 + 后续 auto_resumed 推断（2026-10-29 + # 实弹）；记一条 worker_shutdown_interrupted， + # detail 为任务 id，与 sweep_interrupted 同口径。 + try: + _conn = _db_connect(self.db_path) + try: + _row = _conn.execute( + "SELECT project_id FROM tasks WHERE id = ?", + (task_id,)).fetchone() + finally: + _conn.close() + _pid = _row[0] if _row is not None else "" + if _pid is None: + _pid = "" + except Exception: + _pid = "" + self._audit(_pid, "worker_shutdown_interrupted", + task_id) return all_results diff --git a/host.py b/host.py index 2b4582a..0495a87 100644 --- a/host.py +++ b/host.py @@ -1944,6 +1944,12 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify, # max_rounds=0 是不限轮次：while 循环只在模型宣告、预算耗尽 # 或取消时结束；非 0 时按旧语义穷尽轮次即 unmet。 round_no = int(start_round) - 1 + # 连续失败轮次护栏：回合瞬间失败（如密钥缺失）时不限轮次 + # 循环会无限空转（2026-10-29 实弹：断点后 274 轮全失败， + # state 文件持续增长）。3 轮连续失败即走中断出口，任务 + # 记 interrupted、可断点续跑，不再原地空转；偶发单轮失败 + # 不受影响（计数在成功轮清零）。 + consecutive_failed = 0 while True: round_no += 1 if int(max_rounds) != 0 and round_no > int(max_rounds): @@ -2002,6 +2008,13 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify, goal_usage = _add_usage(goal_usage, result.get("usage")) last_turn_status = result["status"] last_turn_text = result.get("text", "") or "" + if last_turn_status == "failed": + consecutive_failed += 1 + if consecutive_failed >= 3: + return _interrupt_loop( + Exception("连续 3 轮回合失败，停止空转")) + else: + consecutive_failed = 0 if verify is None: # 新形态没有验收命令可跑，v_rc 记 -（轮次行格式不变， # 云端解析只读 verify_out 做 unmet 详情，不受影响）。 diff --git a/tests/test_host.py b/tests/test_host.py index fecd64e..3e4e16c 100644 --- a/tests/test_host.py +++ b/tests/test_host.py @@ -920,6 +920,33 @@ class GoalLoopLongTaskTest(unittest.TestCase): saved = host.load_state(state_path) self.assertEqual(saved["tasks"]["g"]["goal_rounds_done"], 2) + def test_goal_consecutive_failed_turns_stop_spinning(self): + # 不限轮次 + 瞬间失败轮次会无限空转（2026-10-29 实弹： + # 断点后 274 轮全失败）；3 轮连续失败即中断出口。 + class FakeServer: + def __init__(self): + self.calls = 0 + + def run_turn(self, thread_id, prompt, model=None, effort=None, + sandbox=None, timeout=None): + self.calls += 1 + return {"status": "failed", "text": "", + "turn_id": "f%d" % self.calls} + + with tempfile.TemporaryDirectory() as tmp: + state_path = os.path.join(tmp, "host-state.json") + state = {"tasks": {}, "automations": []} + task = self._task(tmp) + state["tasks"]["g"] = task + srv = FakeServer() + result = host.run_goal_loop(srv, state, state_path, + task, "make it", None, 0, "m", + "low", None, cfg={}, timeout=600) + self.assertEqual(srv.calls, 3) + self.assertTrue(result["interrupted"]) + self.assertEqual(task["status"], "interrupted") + self.assertIn("连续 3 轮", task["goal_interrupted_reason"]) + def test_goal_resume_continues_from_checkpoint(self): class FakeServer: def __init__(self):