host
e74f20608b
2026-10-29 live test: task A was interrupted 61s after claim but had never written anything (no host-state entry, no worktree, no trace, no checkpoint); auto-resume ran goal-resume, which host rightly refused, and the task failed. Refusal only makes sense for tasks with progress but a lost breakpoint (those go through the P0 checkpoint rebuild); for a never-started task the only meaningful outcome is a fresh run, identical to its first attempt (same task id, same goal, worktree does not exist yet so nothing is overwritten). Change (claim path only, via _resolve_breakpoint_resume): - rebuild from checkpoint first, unchanged; - if rebuild fails AND _task_never_started holds (state entry, checkpoint row, worktree dir, trace file all absent), clear the breakpoint flag so the run takes the fresh goal path, with a worker_fresh_restart audit event; - any evidence of prior execution, or any check failure, keeps the old refusal path (fail-closed); the auto-resume count cap and no-progress stall guard still bound retries. Tests: cloud/tests/test_neverstart_fresh.py, 10 items (fallback holds, entry/checkpoint/worktree/trace presence each block it, corrupt state blocks it, rebuild still wins when a checkpoint exists); test_goal_resume's breakpoint-routing test now sets up a task with a checkpoint so it still exercises goal-resume. Full suites: cloud 1178 (skipped 4), host 104, deploy 39, green.
209 lines
7.8 KiB
Python
209 lines
7.8 KiB
Python
"""G10 云端 goal 断点续跑单测：端点口径 + worker 断点路由。
|
||
|
||
不跑真实模型回合、不联网：server 侧走本地临时端口 + 临时
|
||
SQLite；worker 侧注入假 runner，只断言命令形状与最终状态。
|
||
本文件不含任何真实密钥、token、证书或私钥。
|
||
"""
|
||
|
||
import http.client
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import unittest
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
try:
|
||
from cloud.auth import AuthStore
|
||
from cloud.server import create_server
|
||
from cloud.worker import Worker
|
||
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.auth import AuthStore
|
||
from cloud.server import create_server
|
||
from cloud.worker import Worker
|
||
|
||
|
||
def _goal_met_runner(cmd):
|
||
return SimpleNamespace(
|
||
returncode=0,
|
||
stdout="round 1: turn=completed verify_rc=- verify_out=\n"
|
||
"goal met after rounds\n",
|
||
stderr="")
|
||
|
||
|
||
class GoalResumeTestCase(unittest.TestCase):
|
||
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.other_id, self.other_key = self.auth.create_user("other")
|
||
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, self.project = self._req(
|
||
"POST", "/projects", {"name": "demo"}, self.owner_key)
|
||
self.assertEqual(status, 201, self.project)
|
||
self.env = {"XCM_API_KEY": "test-placeholder"}
|
||
|
||
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()
|
||
|
||
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 key:
|
||
hdrs["X-API-Key"] = key
|
||
if body is not None:
|
||
payload = json.dumps(body)
|
||
hdrs["Content-Type"] = "application/json"
|
||
conn.request(method, path, body=payload, headers=hdrs)
|
||
resp = conn.getresponse()
|
||
raw = resp.read()
|
||
conn.close()
|
||
try:
|
||
return resp.status, json.loads(raw.decode("utf-8"))
|
||
except ValueError:
|
||
return resp.status, raw.decode("utf-8", errors="replace")
|
||
|
||
def _create(self, **extra):
|
||
body = {"project_id": self.project["id"], "prompt": "build it"}
|
||
body.update(extra)
|
||
return self._req("POST", "/tasks", body, self.owner_key)
|
||
|
||
def _set_status(self, task_id, status):
|
||
conn = sqlite3.connect(self.db_path)
|
||
try:
|
||
conn.execute("UPDATE tasks SET status = ? WHERE id = ?",
|
||
(status, task_id))
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
def _row(self, task_id):
|
||
conn = sqlite3.connect(self.db_path)
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT status, goal_resume_pending FROM tasks"
|
||
" WHERE id = ?", (task_id,)).fetchone()
|
||
return row
|
||
finally:
|
||
conn.close()
|
||
|
||
def test_interrupted_goal_task_resumes_to_queued_with_flag(self):
|
||
status, task = self._create(goal_mode=1)
|
||
self.assertEqual(status, 201, task)
|
||
self._set_status(task["id"], "interrupted")
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/goal-resume" % task["id"],
|
||
{}, self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertEqual(body["status"], "queued")
|
||
self.assertEqual(body["goal_resume_pending"], 1)
|
||
row = self._row(task["id"])
|
||
self.assertEqual(row[0], "queued")
|
||
self.assertEqual(row[1], 1)
|
||
conn = sqlite3.connect(self.db_path)
|
||
try:
|
||
audit = conn.execute(
|
||
"SELECT action FROM audit_log WHERE project_id = ?"
|
||
" AND action = 'task_goal_resumed'",
|
||
(self.project["id"],)).fetchall()
|
||
finally:
|
||
conn.close()
|
||
self.assertEqual(len(audit), 1)
|
||
|
||
def test_terminal_and_failed_and_non_goal_all_409(self):
|
||
status, goal_task = self._create(goal_mode=1)
|
||
self.assertEqual(status, 201, goal_task)
|
||
status, plain_task = self._create()
|
||
self.assertEqual(status, 201, plain_task)
|
||
for task_id, status_value in (
|
||
(goal_task["id"], "completed"),
|
||
(goal_task["id"], "failed"),
|
||
(plain_task["id"], "interrupted")):
|
||
self._set_status(task_id, status_value)
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/goal-resume" % task_id,
|
||
{}, self.owner_key)
|
||
self.assertEqual(status, 409, (task_id, status_value, body))
|
||
row = self._row(task_id)
|
||
self.assertEqual(row[0], status_value)
|
||
self.assertEqual(row[1], 0)
|
||
|
||
def test_non_maintainer_forbidden(self):
|
||
status, task = self._create(goal_mode=1)
|
||
self.assertEqual(status, 201, task)
|
||
self._set_status(task["id"], "interrupted")
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/goal-resume" % task["id"],
|
||
{}, self.other_key)
|
||
self.assertEqual(status, 403, body)
|
||
row = self._row(task["id"])
|
||
self.assertEqual(row[0], "interrupted")
|
||
|
||
def test_worker_breakpoint_uses_goal_resume_without_goal_text(self):
|
||
# 端点置标记后，worker 认领应走 goal-resume（无 --goal-text），
|
||
# 而不是 goal（撞已存在）或 goal-resume --goal-text（开新循环）。
|
||
# main（#86）起未指定 executor 的任务默认成 docker，本地
|
||
# worker 按认领过滤跳过它是正确行为；本测试要的是本地断点
|
||
# 路由，所以显式指定 executor="local"。
|
||
status, task = self._create(goal_mode=1, executor="local")
|
||
self.assertEqual(status, 201, task)
|
||
self._set_status(task["id"], "interrupted")
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/goal-resume" % task["id"],
|
||
{}, self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
# 本测试测的是“有断点的任务走 goal-resume”：先给任务补
|
||
# 一行库断点（模拟跑过一轮），否则按从未启动回落规则，
|
||
# 它会被当成从未启动而走全新 goal（见
|
||
# test_neverstart_fresh.py）。
|
||
conn = sqlite3.connect(self.db_path)
|
||
try:
|
||
conn.execute(
|
||
"INSERT INTO task_checkpoints(task_id, round_no,"
|
||
" thread_id, goal_status, state_sha, verify_out,"
|
||
" created_at) VALUES(?, ?, ?, ?, ?, ?, ?)",
|
||
(task["id"], 1, "th-test", "goal-running", "sha", "", 0))
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
seen = []
|
||
|
||
def fake(cmd):
|
||
seen.append(list(cmd))
|
||
return _goal_met_runner(cmd)
|
||
|
||
worker = Worker(self.db_path, "/fake/host.py", runner=fake,
|
||
env=self.env)
|
||
result = worker.run_one()
|
||
self.assertEqual(result["status"], "completed", result)
|
||
self.assertEqual(len(seen), 1)
|
||
cmd = seen[0]
|
||
self.assertIn("goal-resume", cmd)
|
||
self.assertNotIn("--goal-text", cmd)
|
||
self.assertNotIn("--max-rounds", cmd)
|
||
row = self._row(task["id"])
|
||
self.assertEqual(row[0], "completed")
|
||
self.assertEqual(row[1], 0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|