From 19fd326a730e96a323238848401c13b46ef3e7c5 Mon Sep 17 00:00:00 2001 From: host Date: Sat, 31 Oct 2026 00:35:18 +0800 Subject: [PATCH] feat: attempt ledger + resume preamble (v6) Foundry 'platform restarts with original input' gap, after reading the existing mechanisms: creation idempotency (UNIQUE project_id+idempotency_key), goal checkpoints (task_checkpoints per round, same-thread breakpoint resume with no-progress fuse), merge/cost/notification CAS, and coordinator child reuse already exist. The uncovered segment is non-goal platform-interrupted tasks: auto_resume requeues them fresh in the same worktree after clearing result/error/ interrupted_reason, so the new run gets only the original prompt and cannot tell what the previous attempt had done. This track adds, in worker/server layer only (no host.py or codex thread change): task_attempts ledger, recorded by auto_resume_interrupted in the same transaction before clearing (attempt_no = old auto_resume_count + 1, kind goal_breakpoint|platform_fresh, interrupted_reason, checkpoint_round, result/error truncated to 2000 chars, INSERT OR IGNORE keeps the first record on duplicate); resume preamble injected in-memory only at claim for platform_fresh reruns (previous attempt facts, truncated to 1000 chars, plus check-worktree-first and verify-before- recreating-external-artifacts instructions; skipped for coordinator/_resume/breakpoint tasks and when no record exists, never fabricated); attempts surfaced read-only in GET /tasks/{id}/timeline under existing project-read permission. Fresh PG schema, migrations table list, and server/worker ensure paths share one DDL. Deliberately NOT claimed: the preamble is an instruction, not a hard guarantee like creation idempotency or merge CAS; the ledger does not track worktree files or external artifacts (no worktree snapshot, executor-dependent paths; hard dedup for those still rests on their existing gates); manual retry paths do not go through auto_resume, so they get no ledger record or preamble. Tests: test_attempts.py (7 tests). --- cloud/attempts.py | 169 ++++++++++++++++++++++ cloud/db.py | 12 ++ cloud/migrations.py | 2 + cloud/server.py | 26 +++- cloud/tests/test_attempts.py | 263 +++++++++++++++++++++++++++++++++++ cloud/worker.py | 74 +++++++++- 6 files changed, 541 insertions(+), 5 deletions(-) create mode 100644 cloud/attempts.py create mode 100644 cloud/tests/test_attempts.py diff --git a/cloud/attempts.py b/cloud/attempts.py new file mode 100644 index 0000000..f27ff1b --- /dev/null +++ b/cloud/attempts.py @@ -0,0 +1,169 @@ +"""尝试账本（v6，Foundry“平台以原始输入重启”对应的另一半）。 + +已有的、不在本轨重复造的机制（查代码确认）： +- 任务创建幂等：tasks 的 UNIQUE(project_id, idempotency_key)， + 重复创建回读已有任务，并发抢建靠 IntegrityError 回读。 +- goal 断点：task_checkpoints 每轮快照（worker 镜像线程 + + 终局导入），goal 任务自动续跑走同一线程、同一 worktree、 + 从 goal_rounds_done 下一轮起；无进展熔断（断点轮次 3 次 + 不变即 stall）与退避、次数上限并存。 +- 合并与通知的 CAS：终局写回只在 running 时成功才写结果、 + 记成本、发通知，重复终局不重复记账发通知；协调员子任 + 务按分工名复用，不重复建。 + +本轨补的缺口，恰恰是上述机制都没盖住的那一段：非 goal +任务被平台中断后，auto_resume_interrupted 会把它全新重 +排——同一 worktree 里文件还在，但重排前会清掉 result_text / +error_text / interrupted_reason，新一轮 agent 只拿到原始 +prompt，既不知道上一尝试做到哪一步、出了什么错，也无法 +区分“worktree 里已有的文件是上一尝试的产物还是仓库原有 +内容”，于是可能重复创建上一尝试已创建的东西。goal 任务 +不受此影响（线程连续性自带历史），协调员子任务也不受此 +影响（auto_resume 本来就不接 coordinator_orphan）。 + +本轨只做三件可执行的事，全部在 worker/server 层，不改 +host.py、不改 codex 线程恢复： +① 尝试账本 task_attempts：auto_resume 把任务回 queued + 之前，先把正结束的这次尝试记下来——attempt_no（= 旧 + auto_resume_count + 1，即正结束尝试的序号）、kind + （goal_breakpoint 或 platform_fresh）、interrupted_reason、 + checkpoint_round（库中断点最大轮次，非 goal 恒为 -1）、 + result_text / error_text 各截断 2000 字。记账与回 queued + 在同一事务里，记账失败不单独炸恢复循环（外层已兜底）， + 但同一 (task_id, attempt_no) 重复记时 INSERT OR IGNORE + 保留第一次，不覆盖。 +② 重跑前言：非 goal 平台全新重排的任务在认领时，worker + 在内存副本的 prompt 前贴一段上次尝试说明（原 prompt + 不落库，与项目/待办/记忆注入同款只改内存副本），内容 + 是账本里的事实——第几次尝试、中断原因、断点轮次、上一 + 尝试部分输出与错误（各截断 1000 字）——并明确要求先查 + worktree 现状再动手、对 worktree 外的产物先核查再创建。 + goal 断点续跑、_resume 任务、协调员任务一律跳过（它们 + 已有会话连续性或不跑模型回合）；账本里查不到记录的老 + 任务原样返回，不编造。 +③ 时间线读回：GET /tasks/{id}/timeline 在 steps / + checkpoints 之外加 attempts 数组，项目 read 权限复用 + 时间线原有校验，只读不改。 + +诚实边界（写进模块与交付，不宣称）： +- 前言是指令不是保证：agent 仍可能重复创建，账本只能 + 降低重复概率，不能像创建幂等键、合并 CAS 那样硬保证 + 不重复。worktree 外的产物（分支、PR、发布等）账本不 + 追踪——它只记 DB 里已有的事实，不读 worktree、不调外 + 部 API；那些产物的硬去重仍靠各自已有的门（创建幂等键、 + 合并状态 CAS、分工名复用）。 +- 本轨不做 worktree 文件快照：快照路径因执行器而异 + （local / Docker 租户 / cloud 的 worktree 位置不同）， + 在 worker 层硬做快照会把执行器差异漏进账本；agent 在 + 本轮里自己查 worktree 现状是更可靠的口径。 +- 手动重试（/retry、/continue 等）不经过 auto_resume， + 故不记尝试账本、不贴前言——用户显式重试时由用户自己 + 在新消息里给上下文，本轨不替用户编造上一尝试说明。 +""" + +import time + +ATTEMPT_TEXT_CHARS = 2000 +PREAMBLE_TEXT_CHARS = 1000 + +CREATE_TABLE_SQL = ( + "CREATE TABLE IF NOT EXISTS task_attempts(" + "task_id TEXT, " + "attempt_no INTEGER, " + "kind TEXT, " + "interrupted_reason TEXT, " + "checkpoint_round INTEGER, " + "result_text TEXT, " + "error_text TEXT, " + "created_at DOUBLE PRECISION, " + "PRIMARY KEY(task_id, attempt_no))") + + +def ensure_attempts_table(conn): + """幂等建尝试账本表（server/worker 两支共用同一 DDL）。""" + conn.execute(CREATE_TABLE_SQL) + + +def _trunc(text, limit): + if text is None: + return None + text = str(text) + return text[:limit] if len(text) > limit else text + + +def record_attempt(conn, task_id, attempt_no, kind, + interrupted_reason, checkpoint_round, + result_text, error_text): + """记一次正结束的尝试；重复记同一序号时保留第一次。 + + 调用方负责建表（ensure_attempts_table）与事务提交； + 本函数只执行 INSERT OR IGNORE，不提交、不抛主键冲突。 + """ + conn.execute( + "INSERT OR IGNORE INTO task_attempts(task_id, attempt_no," + " kind, interrupted_reason, checkpoint_round, result_text," + " error_text, created_at) VALUES(?, ?, ?, ?, ?, ?, ?, ?)", + (task_id, attempt_no, kind, interrupted_reason, + checkpoint_round, _trunc(result_text, ATTEMPT_TEXT_CHARS), + _trunc(error_text, ATTEMPT_TEXT_CHARS), time.time())) + + +def latest_attempt(conn, task_id): + """最新一次尝试记录；表缺失或无记录时返回 None，绝不炸。""" + try: + row = conn.execute( + "SELECT task_id, attempt_no, kind, interrupted_reason," + " checkpoint_round, result_text, error_text, created_at" + " FROM task_attempts WHERE task_id = ?" + " ORDER BY attempt_no DESC LIMIT 1", + (task_id,)).fetchone() + except Exception: + return None + if row is None: + return None + try: + return { + "task_id": row[0], "attempt_no": row[1], "kind": row[2], + "interrupted_reason": row[3], "checkpoint_round": row[4], + "result_text": row[5], "error_text": row[6], + "created_at": row[7], + } + except (IndexError, TypeError): + return None + + +def build_resume_preamble(attempt): + """由尝试记录拼重跑前言；无记录或非全新重排时返回空串。 + + 只用账本里的事实，不推测上一尝试做了哪些文件改动 + （账本不记文件清单，见模块注释），核查动作交给本轮 + agent 在 worktree 里做。 + """ + if not isinstance(attempt, dict): + return "" + if attempt.get("kind") != "platform_fresh": + return "" + lines = ["[上次尝试说明：平台中断后重新开始]", + "这是第 %s 次尝试的重新开始，不是全新任务。" + % (attempt.get("attempt_no") or "?"), + "上一尝试中断原因：%s；断点轮次：%s（-1 表示无断点，" + "非 goal 任务恒为 -1）。" + % (attempt.get("interrupted_reason") or "未记录", + attempt.get("checkpoint_round") + if attempt.get("checkpoint_round") is not None + else "未知")] + prev_result = _trunc(attempt.get("result_text"), + PREAMBLE_TEXT_CHARS) + prev_error = _trunc(attempt.get("error_text"), + PREAMBLE_TEXT_CHARS) + if prev_result: + lines.append("上一尝试的部分输出：%s" % prev_result) + if prev_error: + lines.append("上一尝试的错误：%s" % prev_error) + lines.append( + "同一 worktree 里上一尝试的文件仍在：先检查 worktree " + "现状（例如文件列表与改动）再动手，确认哪些已做、哪些 " + "未做；不要重复创建上一尝试已创建的东西；对 worktree " + "外的产物（分支、PR、发布等）无法从 worktree 确认时，" + "先核查再创建，不要假设没创建过。") + return "\n".join(lines) diff --git a/cloud/db.py b/cloud/db.py index 630d1d3..c81ed2a 100644 --- a/cloud/db.py +++ b/cloud/db.py @@ -831,6 +831,18 @@ POSTGRES_SCHEMA = ( "created_at DOUBLE PRECISION)", "CREATE INDEX IF NOT EXISTS idx_task_steps_task" " ON task_steps(task_id, step_no)", + # 尝试账本（v6，见 cloud/attempts.py）：(task_id, attempt_no) + # 唯一；PG 新库一次建齐，老库走 server/worker ensure。 + "CREATE TABLE IF NOT EXISTS task_attempts(" + "task_id TEXT, " + "attempt_no INTEGER, " + "kind TEXT, " + "interrupted_reason TEXT, " + "checkpoint_round INTEGER, " + "result_text TEXT, " + "error_text TEXT, " + "created_at DOUBLE PRECISION, " + "PRIMARY KEY(task_id, attempt_no))", "CREATE TABLE IF NOT EXISTS projects(" "id TEXT PRIMARY KEY, " "name TEXT, " diff --git a/cloud/migrations.py b/cloud/migrations.py index 81a593a..bc922f5 100644 --- a/cloud/migrations.py +++ b/cloud/migrations.py @@ -87,6 +87,8 @@ MIGRATION_TABLES = ( # 记 0，不中断其余表；failure_reason 列两端 schema 一致）。 "resident_watches", "resident_polls", + # 尝试账本（v6）：同款复合主键、无身份列。 + "task_attempts", ) _SKIP_TABLES = frozenset({"kb_fts", "sqlite_sequence"}) diff --git a/cloud/server.py b/cloud/server.py index 929ad4d..0746add 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -61,6 +61,7 @@ from cloud.agent_def import default_agents_dir as _agent_default_dir from cloud.agent_def import list_known_skills as _agent_known_skills from cloud.agent_def import parse_agent_definition as _parse_agent_definition from cloud.agent_def import resolve_registry_agent as _resolve_agent +from cloud.attempts import ensure_attempts_table as _ensure_attempts from cloud.auth import ROLES as _MEMBER_ROLES from cloud.auth import AuthStore from cloud.coordinator import parse_parts as _coord_parse_parts @@ -1717,6 +1718,7 @@ def init_db(db_path): _ensure_uploads_tables(conn) _ensure_task_checkpoints_table(conn) _ensure_task_steps_table(conn) + _ensure_attempts(conn) conn.commit() return conn.execute( @@ -1753,6 +1755,7 @@ def init_db(db_path): _ensure_uploads_tables(conn) _ensure_task_checkpoints_table(conn) _ensure_task_steps_table(conn) + _ensure_attempts(conn) ensure_kill_requests_table(conn) # 常驻监听表（P0 S3 的 SQLite 侧）：启动时建一次， # 热路径（claim/run_due/批量）不再逐条 ensure， @@ -5390,10 +5393,12 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): "followups": [_followup_to_dict(r) for r in rows]}) def _task_timeline(self, user_id, task_id): - """GET /tasks/{id}/timeline（P0 三件其三）：一个任务一张 - 时间线。steps 直读 task_steps（claim/resume/round/final）， - checkpoints 直读 task_checkpoints（每轮断点快照）；旧 - 任务（部署前跑完、无步骤记录）返回空数组，不编造。 + """GET /tasks/{id}/timeline（P0 三件其三 + v6 尝试账本）： + 一个任务一张时间线。steps 直读 task_steps + （claim/resume/round/final），checkpoints 直读 + task_checkpoints（每轮断点快照），attempts 直读 + task_attempts（自动续跑前记下的正结束尝试）；旧任务 + （部署前跑完、无步骤记录）返回空数组，不编造。 """ row = self._fetch_task(task_id) if row is None: @@ -5409,6 +5414,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): try: _ensure_task_checkpoints_table(conn) _ensure_task_steps_table(conn) + _ensure_attempts(conn) step_rows = conn.execute( "SELECT step_no, kind, summary, state_sha, created_at" " FROM task_steps WHERE task_id = ?" @@ -5419,6 +5425,12 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): " verify_out, created_at FROM task_checkpoints" " WHERE task_id = ? ORDER BY round_no", (task_id,)).fetchall() + attempt_rows = conn.execute( + "SELECT attempt_no, kind, interrupted_reason," + " checkpoint_round, result_text, error_text, created_at" + " FROM task_attempts WHERE task_id = ?" + " ORDER BY attempt_no", + (task_id,)).fetchall() finally: conn.close() self._send_json(200, { @@ -5430,6 +5442,12 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): "goal_status": r[2], "state_sha": r[3], "verify_out": r[4], "created_at": r[5]} for r in cp_rows], + "attempts": [{"attempt_no": r[0], "kind": r[1], + "interrupted_reason": r[2], + "checkpoint_round": r[3], + "result_text": r[4], "error_text": r[5], + "created_at": r[6]} + for r in attempt_rows], }) def _task_files(self, user_id, task_id, query): diff --git a/cloud/tests/test_attempts.py b/cloud/tests/test_attempts.py new file mode 100644 index 0000000..2f00d88 --- /dev/null +++ b/cloud/tests/test_attempts.py @@ -0,0 +1,263 @@ +"""尝试账本 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"], "旧输出") diff --git a/cloud/worker.py b/cloud/worker.py index 6814f53..c0cc65f 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -137,6 +137,10 @@ from types import SimpleNamespace from cloud.agent_def import AgentDefinition as _AgentDefinition from cloud.agent_def import format_instructions_header as _agent_header from cloud.agent_def import missing_required_tools as _missing_tools +from cloud.attempts import build_resume_preamble as _resume_preamble +from cloud.attempts import ensure_attempts_table as _ensure_attempts +from cloud.attempts import latest_attempt as _latest_attempt +from cloud.attempts import record_attempt as _record_attempt from cloud.auth import AuthStore from cloud.coordinator import build_report as _coord_build_report from cloud.coordinator import DEFAULT_POLL_INTERVAL as _COORD_POLL_INTERVAL @@ -1025,7 +1029,9 @@ class Worker: " model, interrupted_reason, resume_after," " auto_resume_stall_count," " checkpoint_round_at_last_resume," - " goal_mode, verify_command, goal FROM tasks" + " goal_mode, verify_command, goal," + " result_text, error_text, auto_resume_count" + " FROM tasks" " WHERE status = 'interrupted'" " AND (paused_by_user IS NULL OR paused_by_user = 0)" " AND (interrupted_reason IS NULL OR" @@ -1093,6 +1099,24 @@ class Worker: # 消息，防它劫持新跑的 prompt。 resume_set = ("goal_resume_pending = 0," " resume_message = NULL") + # 尝试账本（v6）：清掉旧结果之前，先把正结束 + # 的这次尝试记下来；记账与回 queued 在同一 + # 事务里提交。attempt_no = 旧 auto_resume_count + # + 1，即正结束尝试的序号；kind 按 goal 与否 + # 分断点续跑与平台全新重排。记账失败不单独 + # 炸恢复循环——外层已兜底，但本轮仍继续回 + # queued（账本缺一条记录好过任务卡住不恢复； + # 认领时查不到账本即不贴前言，不编造）。 + try: + _ensure_attempts(conn) + _record_attempt( + conn, task_id, + (row[14] or 0) + 1, + "goal_breakpoint" if is_goal + else "platform_fresh", + row[5], cp_round, row[12], row[13]) + except Exception: + pass cur = conn.execute( "UPDATE tasks SET status = 'queued', " + resume_set + @@ -1528,6 +1552,12 @@ class Worker: "summary TEXT, " "state_sha TEXT, " "created_at DOUBLE PRECISION)") + # 尝试账本（v6，见 cloud/attempts.py）：worker 直连 + # 老库时也补，与 server ensure 同 DDL。 + try: + _ensure_attempts(conn) + except Exception: + pass conn.execute( "CREATE INDEX IF NOT EXISTS idx_task_steps_task" " ON task_steps(task_id, step_no)") @@ -1920,6 +1950,45 @@ class Worker: task["id"]) return new_task + def _apply_resume_preamble(self, task): + """重跑前言注入（v6，见 cloud/attempts.py）：只改内存副本。 + + 跳过：协调员任务、_resume 任务、断点续跑任务（已有 + 会话连续性）、auto_resume_count 为 0 的首轮任务、账本 + 里查不到记录的老任务、最新尝试不是 platform_fresh + 的任务（goal 断点续跑走线程连续性，不贴前言）。 + 贴法与项目注入同款：前言 + 原 prompt，goal 任务若 + 意外走到这里也同步贴 goal（执行器取 goal 优先于 + prompt）；查账本失败时原样返回，不炸认领。 + """ + if task.get("coordinator_parts"): + return task + if task.get("_resume") or task.get("_goal_resume_breakpoint"): + return task + if not task.get("auto_resume_count"): + return task + try: + conn = _db_connect(self.db_path) + try: + attempt = _latest_attempt(conn, task["id"]) + finally: + conn.close() + except Exception: + return task + preamble = _resume_preamble(attempt) + if not preamble: + return task + new_task = dict(task) + new_task["prompt"] = preamble + "\n\n" + ( + task.get("prompt") or "") + if task.get("goal_mode") or task.get("verify_command"): + new_task["goal"] = preamble + "\n\n" + ( + task.get("goal") or task.get("prompt") or "") + new_task["_resume_preamble_injected"] = True + self._audit(task["project_id"], "worker_resume_preamble_injected", + task["id"]) + return new_task + def _tree_snapshot_source(self, task): """首轮目录树快照的宿主侧来源目录，或 None。 @@ -5742,6 +5811,9 @@ class Worker: # _apply_project_injection 的跳过规则；只改内存副本； # 最后贴，排在待办与记忆之前（最通用上下文在最前）。 task = self._apply_project_injection(task) + # 重跑前言：平台全新重排的任务在项目注入之后贴，离原 + # prompt 最近；只改内存副本，原 prompt 不落库。 + task = self._apply_resume_preamble(task) self._audit(task["project_id"], "worker_claim", "%s executor=%s" % (task_id, self._executor_name())) # log_event 只接受 task_id/project_id/status/message 四个