diff --git a/cloud/sandbox.py b/cloud/sandbox.py index 32de206..5a1b0db 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -117,6 +117,19 @@ def _parse_goal_output(stdout): return None, None +def _goal_resume_refusal(stdout): + """host.py goal-resume 的拒绝行（断点丢失时），无则返回 None。 + + 断点丢失不是可续跑的中断：host-state 里没有任务，任何一次 + 续跑都会原样失败；调用方必须判 failed 终态，绝不能判 + interrupted——否则 worker 自动恢复会 12 秒一轮空转到上限。 + """ + for line in (stdout or "").splitlines(): + if line.startswith("goal-resume 拒绝："): + return line.strip() + return None + + def _goal_interrupted_reason(stdout): """goal interrupted 标记行的原因，无标记返回 None。 @@ -569,8 +582,13 @@ class LocalExecutor(Executor): stderr = getattr(result, "stderr", "") or "" met, goal_error = _parse_goal_output(stdout_full) interrupted_reason = _goal_interrupted_reason(stdout_full) + refusal = _goal_resume_refusal(stdout_full) combined = stdout_full + stderr - if "已存在" in combined: + if refusal is not None and met is None: + # 断点丢失拒绝：终态 failed，不进自动恢复（见 helper）。 + met, goal_error = None, refusal + status = "failed" + elif "已存在" in combined: met, goal_error = None, ( "host 任务已存在：goal 失败请用继续而非重试——" "同任务重试会撞已存在 host-state，worker 会先" @@ -1388,7 +1406,12 @@ class DockerExecutor(Executor): stderr = getattr(result, "stderr", "") or "" met, goal_error = _parse_goal_output(stdout_full) interrupted_reason = _goal_interrupted_reason(stdout_full) - if interrupted_reason is not None: + refusal = _goal_resume_refusal(stdout_full) + if refusal is not None and met is None: + # 断点丢失拒绝：终态 failed，不进自动恢复（见 helper）。 + status = "failed" + goal_error = refusal + elif interrupted_reason is not None: # host 侧已记 interrupted：异常中断不是失败，云端跟 # 着记 interrupted，任务可 /goal-resume 断点续跑。 status = "interrupted" @@ -1484,7 +1507,12 @@ class DockerExecutor(Executor): stderr = getattr(result, "stderr", "") or "" met, goal_error = _parse_goal_output(stdout_full) interrupted_reason = _goal_interrupted_reason(stdout_full) - if interrupted_reason is not None: + refusal = _goal_resume_refusal(stdout_full) + if refusal is not None and met is None: + # 断点丢失拒绝：终态 failed，不进自动恢复（见 helper）。 + status = "failed" + goal_error = refusal + elif interrupted_reason is not None: # host 侧已记 interrupted：异常中断不是失败，云端跟 # 着记 interrupted，任务可 /goal-resume 断点续跑。 status = "interrupted" diff --git a/cloud/tests/test_queue_sandbox.py b/cloud/tests/test_queue_sandbox.py index cfcdaf0..cdddcfa 100644 --- a/cloud/tests/test_queue_sandbox.py +++ b/cloud/tests/test_queue_sandbox.py @@ -391,6 +391,37 @@ class TestGoalExecutors(unittest.TestCase): self.assertEqual(result["status"], "interrupted") self.assertIn("boom", result["goal_error"]) + def test_goal_resume_refusal_is_failed_not_interrupted(self): + # 断点丢失拒绝（2026-10-29 实弹）：host-state 里没有任务时 + # goal-resume 只打印拒绝行、正常退出；三处解析（本地、 + # Docker goal_resume_task、Docker 断点）都必须判 failed + # 终态，绝不能判 interrupted——否则自动恢复 12 秒一轮 + # 空转到 50 次上限。 + refusal = ("goal-resume 拒绝：host-state 里没有该任务 t1，" + "断点已丢失，不能续跑\n") + fake = SimpleNamespace(returncode=0, stdout=refusal, stderr="") + result = LocalExecutor("/fake/host.py", + runner=lambda cmd: fake + )._goal_result_from_runner("t1", fake) + self.assertEqual(result["status"], "failed") + self.assertIn("goal-resume 拒绝", result["goal_error"]) + + def fake_runner(cmd, timeout=None, env=None): + return SimpleNamespace(returncode=0, stdout=refusal, + stderr="") + + with mock.patch("sandbox.shutil.which", + return_value="/usr/bin/docker"): + executor = DockerExecutor( + "/fake/host.py", runner=fake_runner, + env={"XCM_API_KEY": "placeholder-not-a-real-key"}) + for method in ("goal_resume_task", + "goal_resume_breakpoint_task"): + result = getattr(executor, method)( + {"id": "t1", "prompt": "do it"}) + self.assertEqual(result["status"], "failed") + self.assertIn("goal-resume 拒绝", result["goal_error"]) + def test_local_goal_task_already_exists_names_retry(self): def fake(cmd): return SimpleNamespace( diff --git a/host.py b/host.py index 0495a87..177050c 100644 --- a/host.py +++ b/host.py @@ -3079,6 +3079,16 @@ def main(): timeout=args.turn_timeout, verify_timeout=args.verify_timeout) elif args.cmd == "goal-resume": + # 断点丢失守卫（2026-10-29 长任务实弹）：host-state 里没有 + # 该任务时直接拒绝、正常退出，绝不能让 KeyError traceback + # 冲出去——worker 会把无结论的异常退出判成 interrupted， + # 自动恢复再排队、再 KeyError，12 秒一轮空转到 50 次上限 + # 才停。拒绝行由 worker 解析成 failed 终态（断点已丢， + # 续跑无意义），不进自动恢复循环。 + if args.task not in (state.get("tasks") or {}): + print(f"goal-resume 拒绝：host-state 里没有该任务 " + f"{args.task}，断点已丢失，不能续跑") + return t = state["tasks"][args.task] # P1 goal-resume 契约（与 cloud/sandbox.py、cloud/worker.py # 三处一致，绝不能混用）： diff --git a/tests/test_host_resume_missing.py b/tests/test_host_resume_missing.py new file mode 100644 index 0000000..9f7717c --- /dev/null +++ b/tests/test_host_resume_missing.py @@ -0,0 +1,66 @@ +"""host.py goal-resume 断点丢失守卫：state 里没有任务时拒绝不炸。 + +2026-10-29 长任务实弹：断点续跑在 state 里找不到任务时抛 +KeyError traceback，worker 把无结论异常退出判成 interrupted， +自动恢复再排队、再 KeyError，12 秒一轮空转到 50 次上限。 +守卫后：打印拒绝行、正常返回，state 文件不动、线程不恢复。 +""" + +import io +import json +import os +import sys +import tempfile +import unittest +from contextlib import redirect_stdout +from pathlib import Path +from unittest import mock + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT)) + +import host + + +class FakeServer(object): + def __init__(self): + self.resumed = [] + + def initialize(self): + pass + + def close(self): + pass + + def resume_thread(self, thread_id): + self.resumed.append(thread_id) + + +class GoalResumeMissingTest(unittest.TestCase): + def test_missing_task_refuses_without_traceback(self): + with tempfile.TemporaryDirectory() as tmp: + state_path = os.path.join(tmp, "host-state.json") + cfg_path = os.path.join(tmp, "host-config.json") + with open(state_path, "w", encoding="utf-8") as f: + json.dump({"tasks": {"g": {"name": "g", + "thread_id": "x", + "status": "goal-running"}}, + "automations": []}, f) + with open(cfg_path, "w", encoding="utf-8") as f: + json.dump({}, f) + fake = FakeServer() + argv = ["host.py", "--state", state_path, + "--config", cfg_path, + "goal-resume", "--task", "missing-task"] + out = io.StringIO() + with mock.patch.object(sys, "argv", argv): + with mock.patch.object(host, "Server", + lambda **kw: fake): + with redirect_stdout(out): + host.main() + self.assertIn("goal-resume 拒绝", out.getvalue()) + self.assertIn("missing-task", out.getvalue()) + self.assertEqual(fake.resumed, []) + with open(state_path, encoding="utf-8") as f: + saved = json.load(f) + self.assertEqual(list(saved["tasks"].keys()), ["g"])