feat: tamper-evident audit ledger (hash chain) + dual-column cost view

- Hash-chain ledger (cloud/ledger.py): prev_hash/entry_hash per record,
  SQLite append-only triggers, UNIQUE prev/entry hash with fail-closed
  retry at the transaction boundary; verify/export (incl. expected-head
  check for tail truncation); legacy audit_log rows marked pre-chain via
  a single ledger_genesis marker, never rewritten.
- AuthStore.audit() writes audit_log + chain entry in one transaction.
- Track-3 interface: audit_verdict / audit_pushback_round /
  audit_rebuild / third-failure assessment / test-change audit /
  meta-review / rebuild-notice entries; per-(module, contract-version)
  failure counts derived from the ledger (pass or completed rebuild
  resets; total > 3 triggers rebuild); acceptance-triggered rebuilds
  require a confirmed user notice.
- Admin read-only endpoints: GET /admin/audit-ledger/verify|export.
- Cost: dual_column_view (priced_usd_micros + billing note, wording
  reused verbatim from PR #212) wired into project/task cost endpoints;
  by-model and per-creator reporting stay in #212, not duplicated here.
- PG schema + migration lists updated for audit_ledger (incl. sequence
  setval alignment).

cloud/tests/test_tamper_ledger.py

@@ -0,0 +1,596 @@
+"""防篡改账本（哈希链）+ 成本双列视图单测。
+
+不联网、不跑真实模型：账本走临时 SQLite 文件，篡改/删除
+用 SQL 直写模拟（先 DROP 触发器，模拟有 DDL 权限的攻击者；
+触发器本身另有单测），服务端点走本地临时端口。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.cost import CostStore, dual_column_view, price_usd_micros
+    from cloud.ledger import (
+        GENESIS_HASH, LedgerStore, append_pushback_round,
+        audit_timeline, export_chain, list_pushback_rounds,
+        module_failure_counts, verify_chain, verify_export)
+    from cloud.server import create_server, init_db
+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.cost import CostStore, dual_column_view, price_usd_micros
+    from cloud.ledger import (
+        GENESIS_HASH, LedgerStore, append_pushback_round,
+        audit_timeline, export_chain, list_pushback_rounds,
+        module_failure_counts, verify_chain, verify_export)
+    from cloud.server import create_server, init_db
+
+
+def _req(port, method, path, key=None):
+    conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10)
+    hdrs = {}
+    if key is not None:
+        hdrs["X-API-Key"] = key
+    try:
+        conn.request(method, path, headers=hdrs)
+        resp = conn.getresponse()
+        data = resp.read()
+    finally:
+        conn.close()
+    try:
+        parsed = json.loads(data.decode("utf-8")) if data else None
+    except ValueError:
+        parsed = None
+    return resp.status, parsed
+
+
+class TestLedgerChain(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.store = LedgerStore(os.path.join(self.tmp.name, "led.db"))
+
+    def tearDown(self):
+        self.store.close()
+        self.tmp.cleanup()
+
+    def test_append_and_verify_ok(self):
+        self.store.append("u1", "p1", "create_task", "t1")
+        self.store.append("u1", "p1", "approve_merge", "t1")
+        result = self.store.verify()
+        self.assertTrue(result["ok"])
+        self.assertEqual(result["count"], 2)
+        self.assertEqual(result["failures"], [])
+        self.assertNotEqual(result["head_hash"], GENESIS_HASH)
+
+    def test_append_only_interface(self):
+        # 接口层面即只追加：没有修改/删除方法（test_auth 同款断言）。
+        self.assertFalse(hasattr(LedgerStore, "delete_entry"))
+        self.assertFalse(hasattr(LedgerStore, "update_entry"))
+        self.assertFalse(hasattr(LedgerStore, "delete"))
+        self.assertFalse(hasattr(LedgerStore, "update"))
+
+    def test_sqlite_triggers_block_update_and_delete(self):
+        self.store.append("u1", "p1", "create_task", "t1")
+        with self.assertRaises(Exception):
+            self.store.conn.execute(
+                "UPDATE audit_ledger SET detail='x' WHERE id=1")
+        with self.assertRaises(Exception):
+            self.store.conn.execute(
+                "DELETE FROM audit_ledger WHERE id=1")
+        self.store.conn.rollback()
+        self.assertTrue(self.store.verify()["ok"])
+
+    def test_tamper_detail_fails_verify(self):
+        self.store.append("u1", "p1", "create_task", "t1")
+        self.store.append("u1", "p1", "approve_merge", "t1")
+        # 模拟有 DDL 权限的攻击者：先卸触发器再改历史。
+        self.store.conn.execute(
+            "DROP TRIGGER IF EXISTS audit_ledger_no_update")
+        self.store.conn.execute(
+            "UPDATE audit_ledger SET detail='forged' WHERE id=1")
+        self.store.conn.commit()
+        result = self.store.verify()
+        self.assertFalse(result["ok"])
+        self.assertTrue(any("tampered" in f for f in result["failures"]))
+
+    def test_delete_middle_breaks_chain(self):
+        for i in range(3):
+            self.store.append("u1", "p1", "act-%d" % i, "d%d" % i)
+        self.store.conn.execute(
+            "DROP TRIGGER IF EXISTS audit_ledger_no_delete")
+        self.store.conn.execute("DELETE FROM audit_ledger WHERE id=2")
+        self.store.conn.commit()
+        result = self.store.verify()
+        self.assertFalse(result["ok"])
+        self.assertTrue(any("broken" in f for f in result["failures"]))
+
+    def test_delete_tail_detected_by_old_head(self):
+        for i in range(3):
+            self.store.append("u1", "p1", "act-%d" % i, "d%d" % i)
+        old_export = self.store.export()
+        old_head = old_export["head_hash"]
+        self.store.conn.execute(
+            "DROP TRIGGER IF EXISTS audit_ledger_no_delete")
+        self.store.conn.execute(
+            "DELETE FROM audit_ledger WHERE id=3")
+        self.store.conn.commit()
+        # 截断后的链自洽（固有局限，如实断言），但旧头对不上。
+        self.assertTrue(self.store.verify()["ok"])
+        self.assertFalse(
+            self.store.verify(expected_head=old_head)["ok"])
+        self.assertFalse(
+            verify_export(self.store.export(),
+                          expected_head=old_head)["ok"])
+
+    def test_export_round_trip_and_tampered_export(self):
+        self.store.append("u1", "p1", "create_task", "t1")
+        self.store.append("u1", "p1", "approve_merge", "t1")
+        data = self.store.export()
+        self.assertTrue(verify_export(data)["ok"])
+        self.assertEqual(data["count"], 2)
+        data["entries"][0]["detail"] = "forged"
+        self.assertFalse(verify_export(data)["ok"])
+        data2 = self.store.export()
+        data2["head_hash"] = "0" * 64
+        self.assertFalse(verify_export(data2)["ok"])
+
+    def test_legacy_migration_marks_pre_chain_without_rewrite(self):
+        # 老 audit_log 行（直接 INSERT，模拟链上线前的历史）。
+        conn = sqlite3.connect(os.path.join(self.tmp.name, "led.db"))
+        try:
+            conn.execute(
+                "CREATE TABLE IF NOT EXISTS audit_log("
+                "id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT,"
+                " project_id TEXT, action TEXT, detail TEXT,"
+                " created_at REAL)")
+            conn.execute(
+                "INSERT INTO audit_log(user_id, project_id, action,"
+                " detail, created_at) VALUES(?,?,?,?,?)",
+                ("u1", "p1", "create_task", "t0", 1000.0))
+            conn.execute(
+                "INSERT INTO audit_log(user_id, project_id, action,"
+                " detail, created_at) VALUES(?,?,?,?,?)",
+                ("u1", "p1", "approve_merge", "t0", 1001.0))
+            conn.commit()
+        finally:
+            conn.close()
+        # 链已空但老表有行：verify 必须 fail-closed，不认账。
+        fresh = LedgerStore(os.path.join(self.tmp.name, "led.db"))
+        try:
+            result = fresh.verify()
+            self.assertFalse(result["ok"])
+            self.assertTrue(any("legacy_not_migrated" in f
+                                for f in result["failures"]))
+            migrated = fresh.migrate_legacy()
+            self.assertTrue(migrated["migrated"])
+            self.assertEqual(
+                migrated["marker"]["action"], "ledger_genesis")
+            after = fresh.verify()
+            self.assertTrue(after["ok"])
+            self.assertIsNotNone(after["genesis_marker"])
+            self.assertEqual(
+                after["genesis_marker"]["legacy_audit_rows"], 2)
+            # 迁移后链已启动，再跑迁移必须拒绝，不补不改。
+            self.assertFalse(fresh.migrate_legacy()["migrated"])
+            # 老行原样未动：仍 2 行、无哈希列。
+            raw = sqlite3.connect(os.path.join(self.tmp.name, "led.db"))
+            try:
+                rows = raw.execute(
+                    "SELECT COUNT(*) FROM audit_log").fetchone()
+                cols = [c[1] for c in raw.execute(
+                    "PRAGMA table_info(audit_log)").fetchall()]
+            finally:
+                raw.close()
+            self.assertEqual(rows[0], 2)
+            self.assertNotIn("entry_hash", cols)
+        finally:
+            fresh.close()
+
+    def test_auth_audit_writes_chain_same_transaction(self):
+        db_path = os.path.join(self.tmp.name, "auth.db")
+        auth = AuthStore(db_path)
+        try:
+            owner_id, _key = auth.create_user("owner")
+            auth.create_project("p1", owner_id)
+            auth.audit(owner_id, "p1", "create_task", "task-1")
+            auth.audit(owner_id, "p1", "approve_merge", "task-1")
+            self.assertEqual(len(auth.list_audit("p1")), 2)
+        finally:
+            auth.conn.close()
+        led = LedgerStore(db_path)
+        try:
+            result = led.verify()
+            self.assertTrue(result["ok"])
+            self.assertEqual(result["count"], 2)
+            actions = [e["action"] for e in
+                       led.export()["entries"]]
+            self.assertEqual(actions, ["create_task", "approve_merge"])
+        finally:
+            led.close()
+
+
+class TestPushbackAndRebuild(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.store = LedgerStore(os.path.join(self.tmp.name, "led.db"))
+
+    def tearDown(self):
+        self.store.close()
+        self.tmp.cleanup()
+
+    def _verdict(self, verdict, source, task="task-1"):
+        return self.store.append_audit_verdict(
+            task, "p1", "mod-a", "v1", verdict, source,
+            failures=["f1"] if verdict == "fail" else [])
+
+    def test_failure_counts_split_sources_and_reset_on_pass(self):
+        self._verdict("fail", "audit")
+        self._verdict("fail", "acceptance")
+        counts = self.store.module_failure_counts("mod-a", "v1")
+        self.assertEqual(counts["audit_fails"], 1)
+        self.assertEqual(counts["acceptance_fails"], 1)
+        self.assertEqual(counts["total_fails"], 2)
+        self.assertFalse(counts["rebuild_triggered"])
+        self._verdict("pass", "audit")
+        counts = self.store.module_failure_counts("mod-a", "v1")
+        self.assertEqual(counts["total_fails"], 0)
+
+    def test_contract_version_change_restarts_count(self):
+        self._verdict("fail", "audit")
+        self._verdict("fail", "audit")
+        self.assertEqual(
+            self.store.module_failure_counts("mod-a", "v1")["total_fails"],
+            2)
+        self.assertEqual(
+            self.store.module_failure_counts("mod-a", "v2")["total_fails"],
+            0)
+
+    def test_rebuild_trigger_and_rebuild_resets_baseline(self):
+        for source in ("audit", "acceptance", "audit"):
+            self._verdict("fail", source)
+        counts = self.store.module_failure_counts("mod-a", "v1")
+        self.assertEqual(counts["total_fails"], 3)
+        self.assertFalse(counts["rebuild_triggered"])
+        self._verdict("fail", "audit")  # 第 4 次：推倒触发点
+        counts = self.store.module_failure_counts("mod-a", "v1")
+        self.assertTrue(counts["rebuild_triggered"])
+        self.store.append_rebuild_decision(
+            "task-1", "p1", "mod-a", "v1", 1,
+            {"commit": "abc123"}, root_cause="implementation",
+            status="decided", prior_failure_total=4)
+        with self.assertRaises(ValueError):
+            # rebuilt 必须带新实现引用。
+            self.store.append_rebuild_decision(
+                "task-1", "p1", "mod-a", "v1", 1,
+                {"commit": "abc123"}, root_cause="implementation",
+                status="rebuilt")
+        self.store.append_rebuild_decision(
+            "task-1", "p1", "mod-a", "v1", 1,
+            {"commit": "abc123"}, new_impl_ref={"commit": "def456"},
+            root_cause="implementation", status="rebuilt",
+            prior_failure_total=4)
+        counts = self.store.module_failure_counts("mod-a", "v1")
+        self.assertEqual(counts["total_fails"], 0)
+        # 重做后再超 3 次：同样触发（不再自动重开是升级语义，
+        # 账本只负责计数与触发信号，不替第 3 轨做升级决策）。
+        for _ in range(4):
+            self._verdict("fail", "audit")
+        self.assertTrue(
+            self.store.module_failure_counts(
+                "mod-a", "v1")["rebuild_triggered"])
+
+    def test_contract_design_root_cause_recorded_not_rebuild(self):
+        entry = self.store.append_rebuild_decision(
+            "task-1", "p1", "mod-a", "v1", 1,
+            {"tag": "v1-old"}, root_cause="contract_design",
+            status="decided")
+        self.assertEqual(entry["action"], "audit_rebuild")
+        with self.assertRaises(ValueError):
+            self.store.append_rebuild_decision(
+                "task-1", "p1", "mod-a", "v1", 2,
+                {"commit": "x"}, root_cause="something_else")
+
+    def test_timeline_reconstructs_full_path_without_erasure(self):
+        # fail → 推回 1-3 轮 → 第 4 次 fail → 推倒 → 重做 → 再审 pass
+        self._verdict("fail", "audit")
+        for round_no in (1, 2, 3):
+            self.store.append_pushback_round(
+                "task-1", "p1", round_no, ["f1"],
+                {"failures": ["f1"]},
+                module_id="mod-a", contract_version="v1",
+                failure_source="audit")
+            self._verdict("fail", "audit")
+        # 此时累计：首 fail + 3 轮后各 1 fail = 4 次，已触发推倒。
+        self.assertTrue(
+            self.store.module_failure_counts(
+                "mod-a", "v1")["rebuild_triggered"])
+        self.store.append_rebuild_decision(
+            "task-1", "p1", "mod-a", "v1", 1,
+            {"commit": "old"}, new_impl_ref={"commit": "new"},
+            root_cause="implementation", status="rebuilt",
+            prior_failure_total=4)
+        self._verdict("pass", "audit")
+        timeline = self.store.audit_timeline(task_id="task-1")
+        actions = [e["action"] for e in timeline["entries"]]
+        self.assertEqual(actions.count("audit_verdict"), 5)
+        self.assertEqual(actions.count("audit_pushback_round"), 3)
+        self.assertEqual(actions.count("audit_rebuild"), 1)
+        # 最终 pass 在末尾，前面 4 次 fail 一个不少。
+        self.assertEqual(timeline["entries"][-1]["parsed"]["verdict"],
+                         "pass")
+        self.assertEqual(timeline["summary"]["verdict_fail"], 4)
+        self.assertEqual(timeline["summary"]["verdict_pass"], 1)
+        self.assertEqual(timeline["summary"]["rebuilds_completed"], 1)
+        self.assertTrue(self.store.verify()["ok"])
+
+    def test_pushback_round_over_3_flagged_in_timeline(self):
+        self.store.append_pushback_round(
+            "task-1", "p1", 4, ["f1"], {"failures": ["f1"]},
+            module_id="mod-a", contract_version="v1")
+        timeline = self.store.audit_timeline(task_id="task-1")
+        self.assertEqual(len(timeline["summary"]["anomalies"]), 1)
+
+    def test_pushback_rounds_list_and_validation(self):
+        append_pushback_round(
+            self.store.conn, "task-1", "p1", 1, ["f1"],
+            {"failures": ["f1"], "evidence": "e"})
+        self.store.conn.commit()
+        rounds = list_pushback_rounds(self.store.conn, task_id="task-1")
+        self.assertEqual(len(rounds), 1)
+        self.assertEqual(rounds[0]["pushback"]["round_no"], 1)
+        with self.assertRaises(ValueError):
+            append_pushback_round(
+                self.store.conn, "task-1", "p1", 1, [],
+                {"failures": []})
+
+
+class TestTightenedCaliber(unittest.TestCase):
+    """三条收紧：第三次失败判定、保留资产、测试单审、
+    审计的审计、重开前用户确认。"""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.store = LedgerStore(os.path.join(self.tmp.name, "led.db"))
+
+    def tearDown(self):
+        self.store.close()
+        self.tmp.cleanup()
+
+    def _verdict(self, verdict, source="audit", task="task-1"):
+        return self.store.append_audit_verdict(
+            task, "p1", "mod-a", "v1", verdict, source,
+            failures=["f1"] if verdict == "fail" else [])
+
+    def test_third_failure_assessment_kinds(self):
+        entry = self.store.append_third_failure_assessment(
+            "task-1", "p1", "mod-a", "v1", "early_rebuild",
+            {"audit_conclusion": "structural",
+             "failure_items": ["f1"],
+             "diff_scope": "mod-a/*",
+             "passed_items": ["a1"],
+             "continuation_counting": "extra rounds count as fails"})
+        self.assertEqual(entry["action"],
+                         "audit_third_failure_assessment")
+        cont = self.store.append_third_failure_assessment(
+            "task-1", "p1", "mod-a", "v1", "continuation_round",
+            {"audit_conclusion": "nearly done"})
+        parsed = json.loads(cont["detail"])
+        self.assertEqual(parsed["max_extra_rounds"], 1)
+        with self.assertRaises(ValueError):
+            self.store.append_third_failure_assessment(
+                "task-1", "p1", "mod-a", "v1", "early_rebuild",
+                {"audit_conclusion": "x"}, max_extra_rounds=1)
+        with self.assertRaises(ValueError):
+            self.store.append_third_failure_assessment(
+                "task-1", "p1", "mod-a", "v1", "continuation_round",
+                {})
+        timeline = self.store.audit_timeline(task_id="task-1")
+        self.assertEqual(timeline["summary"]["third_failure_assessments"],
+                         2)
+
+    def test_rebuild_retained_assets(self):
+        entry = self.store.append_rebuild_decision(
+            "task-1", "p1", "mod-a", "v1", 1,
+            {"commit": "old"}, new_impl_ref={"commit": "new"},
+            root_cause="implementation", status="rebuilt",
+            retained_assets={
+                "old_test_set_ref": "tests/mod-a@old",
+                "passed_acceptance_items": ["a1", "a2"],
+                "old_new_impl_mapping": {"old": "new"},
+                "old_tests_status": "pass"})
+        parsed = json.loads(entry["detail"])
+        self.assertEqual(parsed["retained_assets"]["old_tests_status"],
+                         "pass")
+        with self.assertRaises(ValueError):
+            self.store.append_rebuild_decision(
+                "task-1", "p1", "mod-a", "v1", 2,
+                {"commit": "old"}, retained_assets={
+                    "old_tests_status": "mostly"})
+
+    def test_test_change_audit_gate(self):
+        # 纯新增不走重新冻结门。
+        self.store.append_test_change_audit(
+            "task-1", "p1", "mod-a", "v1",
+            {"added": ["test_new"]}, auditor="aud-1",
+            contract_refrozen=False)
+        # 删改不走重新冻结：拒绝写入。
+        with self.assertRaises(ValueError):
+            self.store.append_test_change_audit(
+                "task-1", "p1", "mod-a", "v1",
+                {"deleted": ["test_old"]}, auditor="aud-1",
+                contract_refrozen=False)
+        # 删改 + 重新冻结 + 引用：通过。
+        entry = self.store.append_test_change_audit(
+            "task-1", "p1", "mod-a", "v1",
+            {"modified": ["test_m"], "deleted": ["test_old"]},
+            auditor="aud-1", contract_refrozen=True,
+            refreeze_ref="v2", reason="contract change")
+        self.assertEqual(entry["action"], "audit_test_change")
+        with self.assertRaises(ValueError):
+            # refrozen 时引用必填。
+            self.store.append_test_change_audit(
+                "task-1", "p1", "mod-a", "v1",
+                {"modified": ["test_m"]}, auditor="aud-1",
+                contract_refrozen=True)
+
+    def test_meta_review_correction_changes_effective_count(self):
+        first = self._verdict("fail")
+        self._verdict("fail")
+        third = self._verdict("fail")
+        self.assertEqual(
+            self.store.module_failure_counts(
+                "mod-a", "v1")["total_fails"], 3)
+        # 纠正最新一条 fail → pass：计数清零（pass 即基线）。
+        self.store.append_audit_of_audit(
+            "task-1", "p1", "mod-a", "v1", third["id"], "fail",
+            "pass", True, reviewer="meta-1",
+            findings=["evidence misread"], triggered_process="none")
+        self.assertEqual(
+            self.store.module_failure_counts(
+                "mod-a", "v1")["total_fails"], 0)
+        # 但链上原 fail 与纠正都保留，时间线不抹。
+        timeline = self.store.audit_timeline(task_id="task-1")
+        self.assertEqual(timeline["summary"]["verdict_fail"], 3)
+        self.assertEqual(timeline["summary"]["meta_reviews"], 1)
+        self.assertTrue(self.store.verify()["ok"])
+
+    def test_meta_review_first_fail_correction_semantics(self):
+        first = self._verdict("fail")
+        self._verdict("fail")
+        self._verdict("fail")
+        # 纠正第一条 fail → pass：pass 清零后，后两条 fail 重计为 2。
+        self.store.append_audit_of_audit(
+            "task-1", "p1", "mod-a", "v1", first["id"], "fail",
+            "pass", True, reviewer="meta-1",
+            findings=["wrong item"], triggered_process="re-review")
+        self.assertEqual(
+            self.store.module_failure_counts(
+                "mod-a", "v1")["total_fails"], 2)
+
+    def test_acceptance_rebuild_requires_confirmed_notice(self):
+        # 无通知：拒绝。
+        with self.assertRaises(ValueError):
+            self.store.append_rebuild_decision(
+                "task-1", "p1", "mod-a", "v1", 1,
+                {"commit": "old"}, root_cause="implementation",
+                trigger_source="acceptance")
+        # 仅 pending 通知：仍拒绝。
+        self.store.append_rebuild_notice(
+            "task-1", "p1", "mod-a", "v1", 1,
+            "new direction: rebuild mod-a", notified_party="user-1")
+        with self.assertRaises(ValueError):
+            self.store.append_rebuild_decision(
+                "task-1", "p1", "mod-a", "v1", 1,
+                {"commit": "old"}, root_cause="implementation",
+                trigger_source="acceptance")
+        # confirmed 后：通过；audit 触发本来就不走此门。
+        self.store.append_rebuild_notice(
+            "task-1", "p1", "mod-a", "v1", 1,
+            "new direction: rebuild mod-a", notified_party="user-1",
+            confirmation_status="confirmed", confirmed_by="user-1")
+        entry = self.store.append_rebuild_decision(
+            "task-1", "p1", "mod-a", "v1", 1,
+            {"commit": "old"}, root_cause="implementation",
+            trigger_source="acceptance")
+        self.assertEqual(entry["action"], "audit_rebuild")
+        self.store.append_rebuild_decision(
+            "task-1", "p1", "mod-a", "v1", 2,
+            {"commit": "old"}, root_cause="implementation",
+            trigger_source="audit")
+        timeline = self.store.audit_timeline(task_id="task-1")
+        self.assertEqual(timeline["summary"]["rebuild_notices"], 2)
+
+
+class TestCostDualColumn(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.store = CostStore(os.path.join(self.tmp.name, "cost.db"))
+
+    def tearDown(self):
+        try:
+            self.store._conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def test_unpriced_never_priced_as_zero(self):
+        priced = price_usd_micros("gpt-6.1-sol", 1000000, 0, 0)
+        self.assertIsNotNone(priced)
+        self.store.record("t1", "p1", "gpt-6.1-sol",
+                          input_tokens=1000000, output_tokens=0,
+                          usd_micros=priced, source="actual")
+        # 无价模型：usd_micros=None，绝不写成 0。
+        self.store.record("t1", "p1", "unpriced-model",
+                          input_tokens=5000, output_tokens=0,
+                          usd_micros=None, source="actual")
+        view = dual_column_view(self.store.report_by_task("t1"))
+        self.assertEqual(view["priced_usd_micros"], priced)
+        self.assertEqual(view["usd_micros"], priced)
+        self.assertEqual(view["unpriced_actual_tokens"], 5000)
+        self.assertIn("billing_note", view)
+        self.assertIn("不等于免费", view["billing_note"])
+
+
+class TestLedgerServerEndpoints(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "srv.db")
+        auth = AuthStore(self.db_path)
+        try:
+            self.admin_id, self.admin_key = auth.create_user("admin")
+            self.user_id, self.user_key = auth.create_user("plain")
+            auth.create_project("p1", self.admin_id)
+        finally:
+            auth.conn.close()
+        init_db(self.db_path)
+        self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS")
+        os.environ["CODEX_GLOBAL_ADMINS"] = self.admin_id
+        self.server = create_server(self.db_path, port=0)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+
+    def tearDown(self):
+        try:
+            self.server.shutdown()
+            self.server.server_close()
+        except Exception:
+            pass
+        if self._old_admins is None:
+            os.environ.pop("CODEX_GLOBAL_ADMINS", None)
+        else:
+            os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins
+        self.tmp.cleanup()
+
+    def test_verify_and_export_admin_only(self):
+        status, body = _req(self.port, "GET",
+                            "/admin/audit-ledger/verify",
+                            key=self.admin_key)
+        self.assertEqual(status, 200)
+        self.assertTrue(body["ok"])
+        status, body = _req(self.port, "GET",
+                            "/admin/audit-ledger/export?limit=10",
+                            key=self.admin_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(body["version"], 1)
+        status, _body = _req(self.port, "GET",
+                             "/admin/audit-ledger/verify",
+                             key=self.user_key)
+        self.assertEqual(status, 403)
+        status, _body = _req(self.port, "GET",
+                             "/admin/audit-ledger/export?limit=xx",
+                             key=self.admin_key)
+        self.assertEqual(status, 400)
+
+
+if __name__ == "__main__":
+    unittest.main()