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/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"], "旧输出")