"""尝试账本 v6 单测：账本记读、前言拼装、自动续跑记账、 认领时前言注入、时间线读回。 运行（在仓库根目录下）： python3 -m unittest discover -s cloud/tests -p "test_attempts.py" -v 说明：纯函数部分用内存 SQLite；worker 部分用临时 SQLite + init_db，不跑真实模型、不联网；服务端部分用本地临时 端口 + 临时 SQLite（test_p0_three 同款骨架）。 关键断言：自动续跑前旧结果不清账本就先进账本（回 queued 后旧 result_text 已清、账本里仍在）；重跑前言只 改内存 prompt、不改库里原 prompt；goal 断点任务不贴前 言；账本查不到时不编造；前言是指令不是硬保证（见 cloud/attempts.py 模块注释）。 """ import http.client import json import os import sqlite3 import sys import tempfile import threading import time import unittest from pathlib import Path from types import SimpleNamespace try: from cloud.attempts import build_resume_preamble from cloud.attempts import ensure_attempts_table from cloud.attempts import latest_attempt, record_attempt 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.attempts import build_resume_preamble from cloud.attempts import ensure_attempts_table from cloud.attempts import latest_attempt, record_attempt from cloud.auth import AuthStore from cloud.server import create_server, init_db from cloud.worker import Worker def _attempt(**over): base = {"task_id": "t1", "attempt_no": 1, "kind": "platform_fresh", "interrupted_reason": "worker_shutdown", "checkpoint_round": -1, "result_text": "做到一半：已建 a.py", "error_text": "被平台杀掉", "created_at": 1.0} base.update(over) return base class PureTestCase(unittest.TestCase): def test_preamble_contains_facts(self): text = build_resume_preamble(_attempt()) self.assertIn("第 1 次尝试", text) self.assertIn("worker_shutdown", text) self.assertIn("已建 a.py", text) self.assertIn("worktree", text) self.assertIn("先核查再创建", text) def test_preamble_empty_for_non_fresh(self): self.assertEqual( build_resume_preamble(_attempt(kind="goal_breakpoint")), "") self.assertEqual(build_resume_preamble(None), "") self.assertEqual(build_resume_preamble({}), "") def test_record_latest_and_truncation(self): conn = sqlite3.connect(":memory:") ensure_attempts_table(conn) record_attempt(conn, "t1", 1, "platform_fresh", "worker_shutdown", -1, "x" * 3000, None) record_attempt(conn, "t1", 1, "platform_fresh", "sweep", -1, "second", None) latest = latest_attempt(conn, "t1") # 重复记同一序号保留第一次，不覆盖。 self.assertEqual(latest["interrupted_reason"], "worker_shutdown") self.assertEqual(len(latest["result_text"]), 2000) self.assertIsNone(latest_attempt(conn, "nope")) def _insert_task(db_path, task_id, prompt="做活", status="queued", project_id="p1", **cols): now = time.time() conn = sqlite3.connect(db_path) try: base = {"id": task_id, "project_id": project_id, "prompt": prompt, "status": status, "idempotency_key": None, "depends_on": "[]", "priority": 0, "created_by": "tester", "created_at": now, "updated_at": now, "last_heartbeat": None} base.update(cols) keys = list(base.keys()) conn.execute( "INSERT INTO tasks(%s) VALUES(%s)" % (", ".join(keys), ", ".join("?" * len(keys))), [base[k] for k in keys]) conn.commit() finally: conn.close() def _row(db_path, task_id): conn = sqlite3.connect(db_path) try: row = conn.execute( "SELECT status, prompt, result_text, error_text," " interrupted_reason, auto_resume_count FROM tasks" " WHERE id = ?", (task_id,)).fetchone() finally: conn.close() return {"status": row[0], "prompt": row[1], "result_text": row[2], "error_text": row[3], "interrupted_reason": row[4], "auto_resume_count": row[5]} if row else None class WorkerAttemptsTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") init_db(self.db_path) self.env = {"XCM_API_KEY": "test-placeholder"} def tearDown(self): self.tmp.cleanup() def _worker(self): def _runner(cmd): return SimpleNamespace(returncode=0, stdout="{}", stderr="") return Worker(self.db_path, "/fake/host.py", runner=_runner, env=self.env) def test_auto_resume_records_attempt_before_clearing(self): _insert_task(self.db_path, "t1", status="interrupted", interrupted_reason="worker_shutdown", result_text="做到一半：已建 a.py", error_text="被平台杀掉", auto_resume_count=0) worker = self._worker() resumed = worker.auto_resume_interrupted() self.assertEqual(resumed, ["t1"]) row = _row(self.db_path, "t1") self.assertEqual(row["status"], "queued") self.assertIsNone(row["result_text"]) self.assertIsNone(row["interrupted_reason"]) self.assertEqual(row["auto_resume_count"], 1) conn = sqlite3.connect(self.db_path) try: attempt = latest_attempt(conn, "t1") finally: conn.close() self.assertIsNotNone(attempt) self.assertEqual(attempt["attempt_no"], 1) self.assertEqual(attempt["kind"], "platform_fresh") self.assertEqual(attempt["result_text"], "做到一半：已建 a.py") def test_preamble_injection_memory_only(self): _insert_task(self.db_path, "t1", status="queued", prompt="做活", auto_resume_count=1) conn = sqlite3.connect(self.db_path) try: ensure_attempts_table(conn) record_attempt(conn, "t1", 1, "platform_fresh", "worker_shutdown", -1, "做到一半：已建 a.py", "被平台杀掉") conn.commit() finally: conn.close() worker = self._worker() task = worker.get_task("t1") injected = worker._apply_resume_preamble(task) self.assertIn("上次尝试说明", injected["prompt"]) self.assertIn("已建 a.py", injected["prompt"]) self.assertTrue(injected["prompt"].endswith("做活")) # 库里原 prompt 不变。 self.assertEqual(_row(self.db_path, "t1")["prompt"], "做活") def test_preamble_skipped_for_breakpoint(self): _insert_task(self.db_path, "t1", status="queued", prompt="做活", auto_resume_count=1) conn = sqlite3.connect(self.db_path) try: ensure_attempts_table(conn) record_attempt(conn, "t1", 1, "platform_fresh", "worker_shutdown", -1, "旧输出", None) conn.commit() finally: conn.close() worker = self._worker() task = worker.get_task("t1") task = dict(task) task["_goal_resume_breakpoint"] = True self.assertEqual(worker._apply_resume_preamble(task)["prompt"], "做活") class ServerTimelineTestCase(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.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) _insert_task(self.db_path, "T9", status="completed", project_id=self.project["id"]) conn = sqlite3.connect(self.db_path) try: ensure_attempts_table(conn) record_attempt(conn, "T9", 1, "platform_fresh", "worker_shutdown", -1, "旧输出", "旧错") conn.commit() finally: conn.close() 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) headers = {"Content-Type": "application/json"} if key: headers["X-API-Key"] = key conn.request(method, path, json.dumps(body) if body is not None else None, headers) resp = conn.getresponse() raw = resp.read() conn.close() try: return resp.status, json.loads(raw) except Exception: return resp.status, raw def test_timeline_includes_attempts(self): status, body = self._req( "GET", "/tasks/T9/timeline", key=self.owner_key) self.assertEqual(status, 200) self.assertEqual(len(body["attempts"]), 1) self.assertEqual(body["attempts"][0]["attempt_no"], 1) self.assertEqual(body["attempts"][0]["kind"], "platform_fresh") self.assertEqual(body["attempts"][0]["result_text"], "旧输出")