v2 evidence: bind release evidence with ts/nonce, pin actual-only quota, ledger onboarding checklist, ledger failure outbox, read-only gate verifier (no live wiring)

- cloud/evidence_binding.py: bind/validate/nonce single-use store;
  release endpoint consumes nonce on first validated submission,
  replay -> 409 evidence_replayed, stale/future -> 409
  evidence_stale, partial -> 400, absent -> old path + label.
  Binding blocks exact replay and stale evidence, NOT restamping
  (unsigned evidence cannot prove content freshness).
- cloud/quota.py docstring: actual-only is the v2 integration
  position (already converged in v1 base via fef09ae, NOT in
  main); original PR semantics (#86 all-rows vs #90 actual-only)
  still pending coordinator confirmation at merge time.
- cloud/ledger_onboarding.py: DDL checklist referencing
  PG_DDL/PG_MARKER_DDL constants, read-only verify_ledger_ready
  and ledger_setup_report; default stays OFF, PG DDL stays an
  operator action (never claimed as executed).
- cloud/ledger_outbox.py + worker hook: ledger write failures
  go to a JSONL outbox in addition to the log and can be
  replayed; task outcome still unaffected (rule 7 kept),
  audit fail-closed untouched -- asymmetry documented, not
  silently changed.
- cloud/gate_verification.py: read-only verifier over an
  injected reader (match/mismatch/skipped); live wiring in
  merge/release is NOT possible on the current contract (no
  repo/ref in body or task row, no guaranteed trusted token),
  proven in the module header; no fake verification claimed.
- cloud/tests/test_v2_evidence.py: 18 new tests, all passing.

(cherry picked from commit bd85e33cd5e846fa6ee5d7e2440182dbd9c758b0)

cloud/attempt_hook.py

@@ -210,6 +210,14 @@ POLICY (each rule is enforced in code below, not just prose):
    translate to ``%s`` through ``cloud.db``, DDL uses only
    Postgres-valid types, no autoincrement/identity column),
    exactly like gap 3's PG evidence, and labelled as such.
+   Integration v2 out-of-box closure (see
+   ``cloud/ledger_onboarding.py``): the DDL checklist, the
+   explicit enable path and the read-only
+   ``verify_ledger_ready`` / ``ledger_setup_report`` probe
+   close the "hook off by default + manual PG DDL" gap
+   WITHOUT changing the default (still OFF) or auto-running
+   DDL on production Postgres -- the manual PG step remains
+   an operator action, now checkable instead of assumed.
 
 This module imports ``cloud.attempt_ledger`` only, opens no
 database connection itself (callers pass a connection in), and

cloud/evidence_binding.py

@@ -0,0 +1,155 @@
+"""发布证据的新鲜度与单次使用绑定（集成 v2，代码层）。
+
+首轮缺口（全自动 9.5 扣 0.5）：发布门“重算”保证的是“不用
+旧结论”，不是“证据一定新鲜”——证据体里没有时间戳/nonce
+绑定，调用方把合并时同一份证据体在发布时原样重交，同样
+能通过重算。本模块补这一层，但诚实边界先写清：
+
+- 本绑定防的是**原样重放**（同一 nonce 二次提交）与**陈旧
+  证据**（时间戳超出窗口），不防**重盖戳**：没有签名的证据
+  体，调用方完全可以给旧内容盖一个新时间戳/新 nonce 再交，
+  本模块无法识破。防篡改/防重盖戳需要签名证据或第 5 项的
+  只读求证（向 GitHub 实时核对），那是另一层控制，不在本
+  模块冒充。
+- 绑定是**调用方自愿出示**的：请求体里同时带 evidence_ts
+  与 evidence_nonce 才进入绑定校验；两者都不带时走旧路径
+  （与 #259 契约兼容，响应里标注 evidence_binding=absent，
+  绝不谎称已验证新鲜）。只带其一是格式错误（400），不是
+  “半绑定通过”。把绑定改成强制，是契约决定，须协调员/
+  用户在合轨时定，不在本轨悄悄改 #259。
+- 单次使用语义：nonce 在**首次通过校验的提交**即被消费，
+  无论该次发布最终是 200 还是被其它门（未 merged、推回
+  阻塞）挡下——被挡的调用方下次必须取新证据重盖戳。这
+  是故意的：每次发布尝试都要求当次新鲜证据，而不是“一份
+  证据可反复试到放行”。
+- nonce 存储与审计链无关：evidence_nonces 表只记 nonce、
+  任务 id 与时间戳，不写 audit_log、不移动 #241 链头；409
+  不写审计的规则不受影响。表在写路径（首次消费）才建，
+  读路径（校验已存在）不建表——查不到表时视为无消费记
+  录，不是错误。
+
+零第三方依赖，只用标准库。时间一律用 Unix 秒（time.time），
+与 full_auto.audit_record 的 ts 同口径。
+"""
+
+import os
+import time
+import uuid
+
+# 新鲜度窗口：证据时间戳距提交时刻不得超过 1 小时；未来
+# 容忍 5 分钟时钟偏差，超出即视为异常（不是“更新鲜所以更好”）。
+MAX_AGE_SECONDS = 3600
+CLOCK_SKEW_SECONDS = 300
+
+NONCE_TABLE_DDL = (
+    "CREATE TABLE IF NOT EXISTS evidence_nonces("
+    "nonce TEXT NOT NULL PRIMARY KEY, "
+    "task_id TEXT NOT NULL, "
+    "evidence_ts DOUBLE PRECISION NOT NULL, "
+    "consumed_at DOUBLE PRECISION NOT NULL)"
+)
+
+
+def bind_evidence(evidence, now=None, nonce=None):
+    """给一份证据体盖戳：返回原证据的浅拷贝 + 时间戳/nonce。
+
+    调用方（协调员侧）在**取到证据的那一刻**调用本函数，
+    不要在合并时盖好、发布时原样重交——那样时间戳本身
+    就是旧的，会被窗口校验挡下（这正是本绑定的意义）。
+    nonce 缺省为 uuid4 hex；调用方自带 nonce 时须保证
+    全局唯一（重复 nonce 即重放）。
+    """
+    if not isinstance(evidence, dict):
+        raise ValueError("evidence must be a dict")
+    stamped = dict(evidence)
+    stamped["evidence_ts"] = time.time() if now is None else now
+    stamped["evidence_nonce"] = nonce or uuid.uuid4().hex
+    return stamped
+
+
+def validate_bound_evidence(evidence, now=None):
+    """纯函数校验绑定证据的形状与新鲜度（不查 nonce 库）。
+
+    返回 (ok, reason)：ok 为 False 时 reason 为
+    evidence_binding_missing / evidence_binding_incomplete /
+    evidence_ts_not_a_number / evidence_nonce_empty /
+    evidence_too_old / evidence_from_future。nonce 重放由
+    EvidenceNonceStore.consume 判定，不在本函数（本函数
+    无状态，同一份证据调两次结果相同，那不是重放判定）。
+    """
+    if not isinstance(evidence, dict):
+        return False, "evidence_binding_missing"
+    has_ts = evidence.get("evidence_ts") is not None
+    has_nonce = evidence.get("evidence_nonce") is not None
+    if not has_ts and not has_nonce:
+        return False, "evidence_binding_missing"
+    if not (has_ts and has_nonce):
+        return False, "evidence_binding_incomplete"
+    ts = evidence.get("evidence_ts")
+    if isinstance(ts, bool) or not isinstance(ts, (int, float)):
+        return False, "evidence_ts_not_a_number"
+    nonce = evidence.get("evidence_nonce")
+    if not isinstance(nonce, str) or not nonce.strip():
+        return False, "evidence_nonce_empty"
+    moment = time.time() if now is None else now
+    if ts > moment + CLOCK_SKEW_SECONDS:
+        return False, "evidence_from_future"
+    if moment - ts > MAX_AGE_SECONDS:
+        return False, "evidence_too_old"
+    return True, None
+
+
+def binding_presence(evidence):
+    """绑定出示状态：bound（成对出示）/ partial（只带其一）/
+    absent（两者皆无）。服务端据此决定走绑定校验、报 400
+    还是走旧路径（并标注 absent，绝不谎称已验证）。"""
+    if not isinstance(evidence, dict):
+        return "absent"
+    has_ts = evidence.get("evidence_ts") is not None
+    has_nonce = evidence.get("evidence_nonce") is not None
+    if has_ts and has_nonce:
+        return "bound"
+    if has_ts or has_nonce:
+        return "partial"
+    return "absent"
+
+
+class EvidenceNonceStore:
+    """nonce 单次使用存储（DB 连接由调用方传入并负责提交）。
+
+    consume 成功返回 True（首次见此 nonce）；nonce 已存在
+    返回 False（重放）。表不存在时 consume 先建表（写路径
+    才建）；has() 在表不存在时返回 False，不建表。
+    """
+
+    def __init__(self, conn):
+        self.conn = conn
+
+    def ensure_table(self):
+        self.conn.execute(NONCE_TABLE_DDL)
+
+    def has(self, nonce):
+        try:
+            row = self.conn.execute(
+                "SELECT nonce FROM evidence_nonces WHERE nonce = ?",
+                (nonce,)).fetchone()
+        except Exception:
+            return False
+        return row is not None
+
+    def consume(self, task_id, nonce, evidence_ts, consumed_at=None):
+        """消费一个 nonce：首次 True；已存在 False（不覆盖原行）。
+
+        调用方须在 consume 后提交（本函数不 commit：服务端
+        在自己的连接生命周期里提交，与其余写操作保持一致）。
+        """
+        self.ensure_table()
+        if self.has(nonce):
+            return False
+        self.conn.execute(
+            "INSERT INTO evidence_nonces(nonce, task_id, evidence_ts,"
+            " consumed_at) VALUES(?, ?, ?, ?)",
+            (nonce, task_id, evidence_ts,
+             time.time() if consumed_at is None else consumed_at),
+        )
+        return True

cloud/gate_verification.py

@@ -0,0 +1,140 @@
+"""合并/发布门禁证据的只读核对器（集成 v2，代码层）。
+
+首轮缺口（门禁与审计 9.0 扣 0.5）：硬门禁与发布门禁强制
+的是“调用方提交的证据齐全且全绿”，不是“服务端已向
+GitHub 求证”——调用方谎报 CI 全绿 / 保护状态 / 协调员
+批准时，门禁会放行；docstring 称覆盖控制是“审计与
+PR/Checks 的只读核对”，但在合并/发布路径上没有这样
+一段“调 GitHub Checks API 核对证据”的代码（#210 的
+实时保护检查只属于 auto-config execute 端点，不属于
+merge/release 端点）。
+
+本轨的结论分两半，缺一不可：
+
+A. 核对器本身已实现（本模块，纯函数 + 注入式 reader，
+   有测试）：verify_gate_evidence(evidence, reader) 把
+   调用方证据与 reader 给出的实时数据逐项比对——CI
+   状态、每项 required 检查的状态、分支保护状态；不
+   一致即 mismatch（逐条列出），一致即 verified。reader
+   是调用方注入的 callable：reader(repo, ref) -> {
+   "ci_checks": {name: status}, "branch_protection":
+   "protected"|"unprotected"|... }，真实 reader 用
+   github_app 的 _github_api_request 走 GET Checks /
+   GET 分支保护（只读，绝不写 PR、不改保护、不合并）。
+   reader 为 None 或抛异常时结果是“未核对”
+   （checked=False, verified=False），绝不是“通过”——
+   调用方把 skipped 当 pass 用，是调用方的 bug，本
+   模块的返回值结构让这种误用一眼可见（verified 恒
+   False，除非逐项比对全过）。
+
+B. 实时求证在当前合并/发布路径上**接不上**（不可实现
+   证明，不是未做）：① 请求体里没有 repo 与 ref——
+   merge/release 的证据字段只有 files_changed、ci_checks、
+   branch_protection 等，没有仓库名、没有提交 SHA，
+   reader 不知道向哪个仓库的哪次提交求证，猜仓库
+   就是编造求证；② 服务端没有可信的求证通道保证——
+   github_app.resolve_app_token 对未安装 App 的仓库
+   返回 None（调用方回落 PAT），PAT 是调用方自己的
+   凭证，用它“核对”调用方自己的证据不是独立求证；
+   ③ 任务行里没有 repo/ref 列，服务端无法从任务 id
+   反推求证目标。接上实时求证须先改契约（请求体或任
+   务行加 repo + ref，安装保证或服务端侧可信 token），
+   那是契约决定，须协调员/用户在合轨时定；本轨不编
+   一个“看起来在求证”的假 reader、不把调用方证据
+   自己与自己比对后写成 verified。
+
+当前真实控制（写清，不拔高）：门禁强制证据齐全全绿
+（硬模式 fail-closed）；审计记 outcome 与 blocks
+（merge_gate_blocked / merge_task / release_task），
+但审计 detail 不保存证据体全文——事后追责能查到
+“谁在何时以何 blocks 通过/被挡”，查不到“当时提交
+的 ci_checks 原文”。谎报证据的发现仍靠事后审计与
+GitHub 侧记录比对，不是当场发现。本模块的核对器
+是为契约补齐后的接线准备的：届时服务端只需注入
+真实 reader 并把 verified=False（无论 skipped 还是
+mismatch）当作阻断，核对器本身不用改。
+
+零第三方依赖，只用标准库。本模块不 import cloud
+其它模块（reader 由调用方注入），不发网络请求。
+"""
+
+
+def verify_gate_evidence(evidence, reader=None, repo=None, ref=None):
+    """只读核对调用方门禁证据与实时数据。
+
+    evidence：调用方提交的门禁证据 dict（与硬门禁同款
+    字段：ci_status、ci_checks[{name, status}]、
+    required_checks、branch_protection）。
+    reader：callable (repo, ref) -> {"ci_checks": {名: 状态},
+    "branch_protection": str}，或 None（无可信求证通道）。
+    repo/ref：求证目标；reader 非 None 但 repo/ref 缺失
+    时同样视为未核对（不猜目标）。
+
+    返回 {"checked": bool, "verified": bool, "mismatches": [...],
+    "reason": str|None}：
+    - checked=False：未核对（无 reader、无目标、reader 失败），
+      verified 恒 False，reason 说明原因；调用方绝不能把
+      checked=False 当通过。
+    - checked=True 且 mismatches 为空：逐项比对全过，
+      verified=True。
+    - checked=True 且 mismatches 非空：证据与实时数据不
+      一致，verified=False，mismatches 逐条说明（ci 状态、
+      某项检查、分支保护）。
+    比对规则：只比对调用方声称的项——调用方没声称的
+    检查不判（缺失由硬门禁的证据缺失码处理，不归本
+    模块）；实时数据缺失某项时记 mismatch（声称全绿却
+    查无此项，不是“查不到所以通过”）。
+    """
+    if reader is None:
+        return {"checked": False, "verified": False,
+                "mismatches": [],
+                "reason": "no_trusted_reader"}
+    if not repo or not ref:
+        return {"checked": False, "verified": False,
+                "mismatches": [],
+                "reason": "no_verification_target"}
+    try:
+        live = reader(repo, ref)
+    except Exception as exc:
+        return {"checked": False, "verified": False,
+                "mismatches": [],
+                "reason": "reader_failed: %s" % exc}
+    if not isinstance(live, dict):
+        return {"checked": False, "verified": False,
+                "mismatches": [],
+                "reason": "reader_failed: non-dict result"}
+    mismatches = []
+    live_checks = live.get("ci_checks") or {}
+    if not isinstance(live_checks, dict):
+        live_checks = {}
+    evidence = evidence or {}
+    if evidence.get("ci_status") == "success":
+        failed_live = sorted(
+            name for name, status in live_checks.items()
+            if status != "success")
+        if failed_live:
+            mismatches.append(
+                "ci_not_green_live: %s" % ", ".join(failed_live))
+    for check in (evidence.get("ci_checks") or []):
+        if not isinstance(check, dict):
+            continue
+        name = check.get("name")
+        if not name:
+            continue
+        if check.get("status") == "success" \
+                and live_checks.get(name) != "success":
+            mismatches.append(
+                "check_not_green_live: %s" % name)
+    for name in (evidence.get("required_checks") or []):
+        if live_checks.get(name) != "success":
+            mismatches.append(
+                "required_check_not_green_live: %s" % name)
+    claimed_protection = evidence.get("branch_protection")
+    live_protection = live.get("branch_protection")
+    if claimed_protection == "protected" \
+            and live_protection != "protected":
+        mismatches.append(
+            "branch_protection_not_protected_live: %s"
+            % live_protection)
+    return {"checked": True, "verified": not mismatches,
+            "mismatches": mismatches, "reason": None}

cloud/ledger_onboarding.py

@@ -0,0 +1,190 @@
+"""台账钩子开箱闭合方案（集成 v2，代码层）。
+
+首轮缺口（成本与台账 8.5 扣 0.5）：Worker(attempt_ledger_hook
+=False) 默认关，CLI 须显式 --attempt-ledger-hook；生产 PG
+上启用前须先手动执行台账 PG_DDL 与标记表 PG_MARKER_DDL
+并验证。默认状态下指标率是 unknown（诚实），不是“已用
+真实台账”——“接通”只在“开 flag + 手动跑 DDL + 验证”
+之后成立，开箱未闭合。
+
+本轨的闭合方案（不改任何已定语义）：
+
+1. 默认行为不变：钩子默认仍关。默认关是 #255 的设计
+   （显式启用才写行，关时行为与 gap 3 完全一致、无新表
+   无新行），不是 bug，不为“开箱即用”悄悄改成默认开。
+   把默认改成开，是用户/协调员在合轨时才能定的语义变
+   更，本轨不做。
+2. 显式开启路径唯一且可验证：Worker(attempt_ledger_hook
+   =True) 或 CLI --attempt-ledger-hook；开启后 run_one
+   按 attempt_hook.py 规则 1-4 写行。本模块的
+   ledger_setup_report() 把“表是否齐、钩子是否开”分开
+   报告：表齐但钩子关 ≠ 就绪，钩子开但表不齐 ≠ 就绪。
+3. DDL 清单不另造：清单直接引用 attempt_ledger.PG_DDL
+   与 attempt_hook.PG_MARKER_DDL 两个常量（本模块不
+   复制 DDL 文本，复制就会漂移）。另含 gap 3 旧表升级句
+   attempt_ledger.OLD_TABLE_ALTER（仅当旧 DDL 建过表时
+   需要；gap 3 未合并，生产尚无旧表，属防御项）。
+4. 校验函数 verify_ledger_ready(conn)：只读探针——查两
+   张表的存在与必需列，不建表、不写行（读路径绝不改
+   schema，与 metrics-report 端点的只读纪律一致）。
+5. 生产 PG 手动 DDL 仍是运维侧动作：本轨无生产 PG、无
+   本地 PG，PG 证据与此前各轨同款——适配器级（? 占位符
+   经 cloud.db 翻译、DDL 仅用 PG 合法类型、无自增列），
+   绝不写成“生产 PG 已执行”。运维执行清单见
+   LEDGER_DDL_CHECKLIST（顺序执行 + 每步验证命令）。
+
+零第三方依赖，只用标准库。conn 约定与 attempt_ledger
+相同：DBAPI 连接，? 占位符（PG 经 cloud.db 翻译）。
+"""
+
+from cloud.attempt_hook import MARKER_DDL as _MARKER_DDL
+from cloud.attempt_hook import PG_MARKER_DDL as _PG_MARKER_DDL
+from cloud.attempt_ledger import OLD_TABLE_ALTER as _OLD_TABLE_ALTER
+from cloud.attempt_ledger import PG_DDL as _LEDGER_PG_DDL
+
+# 必需列：缺任一列即视为表不齐（旧 DDL 建的表缺
+# corrects_attempt_no 时走 OLD_TABLE_ALTER 补列）。
+LEDGER_REQUIRED_COLUMNS = (
+    "task_id", "attempt_no", "outcome", "is_defect",
+    "defect_category", "created_at", "corrects_attempt_no",
+)
+MARKER_REQUIRED_COLUMNS = ("task_id", "origin", "queued_at")
+
+# 运维执行清单（顺序）：每步含“做什么 + 怎么验证”。本
+# 清单是文档数据，不是执行器——生产 PG 上的执行仍由
+# 运维手工完成，本模块不代执行、不连生产库。
+LEDGER_DDL_CHECKLIST = (
+    {
+        "step": 1,
+        "action": "在生产 PG 上执行 attempt_ledger.PG_DDL"
+                  "（建 task_attempts 表）",
+        "ddl_source": "cloud.attempt_ledger.PG_DDL",
+        "verify": "\\d task_attempts 或 information_schema"
+                  " 查表与列齐全",
+    },
+    {
+        "step": 2,
+        "action": "在生产 PG 上执行 attempt_hook.PG_MARKER_DDL"
+                  "（建 task_attempt_origin_markers 表）",
+        "ddl_source": "cloud.attempt_hook.PG_MARKER_DDL",
+        "verify": "\\d task_attempt_origin_markers 或"
+                  " information_schema 查表与列齐全",
+    },
+    {
+        "step": 3,
+        "action": "仅当表由 gap 3 旧 DDL 建过时：执行"
+                  " attempt_ledger.OLD_TABLE_ALTER 补"
+                  " corrects_attempt_no 列（gap 3 未合并，"
+                  "生产正常路径跳过此步）",
+        "ddl_source": "cloud.attempt_ledger.OLD_TABLE_ALTER",
+        "verify": "task_attempts 含 corrects_attempt_no 列",
+    },
+    {
+        "step": 4,
+        "action": "启用钩子：worker 加 --attempt-ledger-hook"
+                  "（或 Worker(attempt_ledger_hook=True)）后重启",
+        "ddl_source": None,
+        "verify": "新派发产生 task_attempts 行；"
+                  "平台重派发不产生计数行（marker 生效）",
+    },
+    {
+        "step": 5,
+        "action": "用 verify_ledger_ready(conn) 复核："
+                  "tables_ready 且 hook_enabled 方为就绪",
+        "ddl_source": None,
+        "verify": "ledger_setup_report(conn, hook_enabled=True)"
+                  " 的 ready 为 True",
+    },
+)
+
+
+def ledger_ddl_sources():
+    """DDL 清单的权威来源（返回常量本身，供运维原样执行
+    与单测钉住“清单与实现同一来源”，防止清单漂移）。"""
+    return {
+        "task_attempts_pg_ddl": _LEDGER_PG_DDL,
+        "marker_pg_ddl": _PG_MARKER_DDL,
+        "old_table_alter": _OLD_TABLE_ALTER,
+        "marker_ddl": _MARKER_DDL,
+    }
+
+
+def _table_columns(conn, table):
+    """只读探针：表不存在返回 None；存在返回列名集合。
+
+    PRAGMA table_info 经 cloud.db 适配层翻译（与
+    CostStore._ensure_cached_column 同款做法）；探针失败
+    （如 PG 适配差异）返回 None，按“不齐”处理，绝不
+    按“齐”冒充。
+    """
+    try:
+        rows = conn.execute(
+            "PRAGMA table_info(%s)" % table).fetchall()
+    except Exception:
+        return None
+    if not rows:
+        return None
+    try:
+        return {row[1] for row in rows}
+    except (IndexError, TypeError, KeyError):
+        return None
+
+
+def verify_ledger_ready(conn):
+    """只读校验两张台账表：返回 {tables_ready, missing}。
+
+    missing 列出缺失项，如 "task_attempts table missing"、
+    "task_attempts missing columns: corrects_attempt_no"、
+    "task_attempt_origin_markers table missing"。本函数不
+    建表、不写行、不提交。
+    """
+    missing = []
+    ledger_cols = _table_columns(conn, "task_attempts")
+    if ledger_cols is None:
+        missing.append("task_attempts table missing")
+    else:
+        absent = [c for c in LEDGER_REQUIRED_COLUMNS
+                  if c not in ledger_cols]
+        if absent:
+            missing.append("task_attempts missing columns: %s"
+                           % ", ".join(absent))
+    marker_cols = _table_columns(conn, "task_attempt_origin_markers")
+    if marker_cols is None:
+        missing.append("task_attempt_origin_markers table missing")
+    else:
+        absent = [c for c in MARKER_REQUIRED_COLUMNS
+                  if c not in marker_cols]
+        if absent:
+            missing.append(
+                "task_attempt_origin_markers missing columns: %s"
+                % ", ".join(absent))
+    return {"tables_ready": not missing, "missing": missing}
+
+
+def ledger_setup_report(conn, hook_enabled=False):
+    """开箱就绪报告：表齐 + 钩子开，两者缺一都不算就绪。
+
+    返回 {ready, tables_ready, hook_enabled, missing,
+    next_steps}：next_steps 是运维下一步动作的文字清单
+    （按 LEDGER_DDL_CHECKLIST 顺序，未做项才列出），
+    不是自动执行。本函数同样只读，不建表、不写行。
+    """
+    check = verify_ledger_ready(conn)
+    enabled = bool(hook_enabled)
+    steps = []
+    if check["missing"]:
+        steps.append("run the PG DDL steps in LEDGER_DDL_CHECKLIST"
+                     " (steps 1-3) on production Postgres and verify")
+    if not enabled:
+        steps.append("enable the hook with --attempt-ledger-hook"
+                     " (Worker(attempt_ledger_hook=True)) and restart")
+    if not steps:
+        steps.append("ledger is live: counted dispatches write rows;"
+                     " rates are real, not unknown")
+    return {
+        "ready": bool(check["tables_ready"] and enabled),
+        "tables_ready": check["tables_ready"],
+        "hook_enabled": enabled,
+        "missing": check["missing"],
+        "next_steps": steps,
+    }

cloud/ledger_outbox.py

@@ -0,0 +1,178 @@
+"""台账写失败的 durable 记录与重放（集成 v2，代码层）。
+
+首轮缺口（成本与台账 8.5 扣 0.5）：_record_ledger_attempt
+写失败只记一条 worker_attempt_ledger_failed 日志，绝不
+影响任务状态与返回值（attempt_hook.py 规则 7）。这是
+设计（指标台账不应挡任务），但与 #241 审计账本的
+fail-closed 不对称：一次写失败会让率悄悄少算一行，只
+有日志可查，没有重放/补写路径。
+
+本轨的对齐方式（写清边界，不悄悄改任一侧语义）：
+
+- 不把台账改成 fail-closed：台账写失败仍不挡任务、不
+  改任务状态、不改返回值（规则 7 原样保留）。让指标写
+  入能阻塞任务执行，是比丢一行更坏的 trade-off，
+  本轨不做。
+- 不把审计改成 best-effort：审计账本 fail-closed 原样
+  保留（审计是决策级证据，丢一行即链不可信；台账是
+  指标，丢一行只影响率的分母/分子，且可事后补）。
+- 不对称的剩余部分用“不丢”补齐：写失败时除日志外，
+  再向 worker 侧 outbox 文件追加一行 JSONL（task_id、
+  outcome、error、failed_at）；运维事后调
+  replay_failures() 把 outbox 行经 record_attempt 补写
+  进台账（append-only，attempt_no 按 MAX+1 续号），
+  成功补写的行从 outbox 清除，仍失败的行保留待下次。
+  于是失败不再是“只有日志可查”，而是“可数、可重放、
+  重放后可验证行数补齐”。
+
+诚实边界：
+- outbox 本身也是 best-effort：outbox 追加失败（如
+  磁盘满）只记日志，不抛、不挡任务——此时仍退回首轮
+  状态（仅日志可查），但这条路径本身已被测试钉住
+  （追加失败不抛异常）。
+- 重放无幂等键：record_attempt 没有去重键；若原始写
+  入实际已提交、只是提交确认丢失（ambiguous commit），
+  重放会多写一行。重放是运维手动动作：重放前须核对
+  task_attempts 是否已有该任务的对应行，确认缺失再
+  放行；本函数不假装能自动去重（去重键是新的 schema
+  语义，须另立一轨定）。
+- outbox 路径：环境变量 CLOUD_LEDGER_FAILURE_OUTBOX
+  显式指定时优先；否则取 worker db_path 同目录下的
+  attempt_ledger_failures.jsonl；db_path 为 :memory:
+  时无落盘位置，返回 None（调用方只记日志）。
+
+零第三方依赖，只用标准库。
+"""
+
+import json
+import os
+import time
+
+OUTBOX_ENV = "CLOUD_LEDGER_FAILURE_OUTBOX"
+OUTBOX_FILENAME = "attempt_ledger_failures.jsonl"
+
+
+def outbox_path_for(db_path, env=None):
+    """解析 outbox 文件路径：显式 env > db 同目录 > None。
+
+    env 缺省读 os.environ（调用方可注入 dict 做单测）。
+    db_path 为 :memory: 或空时无同目录可言，返回 None。
+    """
+    if env is None:
+        env = os.environ
+    explicit = env.get(OUTBOX_ENV) if env else None
+    if explicit:
+        return str(explicit)
+    if not db_path or str(db_path) == ":memory:":
+        return None
+    directory = os.path.dirname(os.path.abspath(str(db_path)))
+    return os.path.join(directory, OUTBOX_FILENAME)
+
+
+def append_failure(path, task_id, outcome, error):
+    """向 outbox 追加一行失败记录。返回 True/False（写成功与否）。
+
+    追加失败返回 False，不抛异常（调用方只记日志）。
+    行格式：{task_id, outcome, error, failed_at}，一行一
+    条 JSON，坏行在重放时跳过并保留（见 replay_failures）。
+    """
+    if not path:
+        return False
+    record = {
+        "task_id": task_id,
+        "outcome": outcome,
+        "error": str(error),
+        "failed_at": time.time(),
+    }
+    try:
+        directory = os.path.dirname(os.path.abspath(str(path)))
+        if directory:
+            os.makedirs(directory, exist_ok=True)
+        with open(path, "a", encoding="utf-8") as handle:
+            handle.write(json.dumps(record, ensure_ascii=False) + "\n")
+        return True
+    except Exception:
+        return False
+
+
+def read_failures(path):
+    """读 outbox：返回 (records, bad_lines)。
+
+    records 为解析成功的 dict 列表（按文件顺序）；bad_lines
+    为解析失败的行数（坏行不丢弃，重放时原样保留）。
+    文件不存在返回 ([], 0)，不是错误。
+    """
+    records = []
+    bad = 0
+    try:
+        with open(path, encoding="utf-8") as handle:
+            lines = handle.readlines()
+    except OSError:
+        return records, bad
+    for line in lines:
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            item = json.loads(line)
+        except ValueError:
+            bad += 1
+            continue
+        if not isinstance(item, dict) or not item.get("task_id") \
+                or item.get("outcome") not in (
+                    "success", "failure", "interrupted"):
+            bad += 1
+            continue
+        records.append(item)
+    return records, bad
+
+
+def replay_failures(path, conn, record_fn=None):
+    """重放 outbox：经 record_attempt 补写，返回统计。
+
+    返回 {replayed, failed, remaining, bad_lines}：
+    replayed 为成功补写行数，failed 为补写仍失败的行数，
+    remaining 为重放后 outbox 剩余行数（failed + bad 行），
+    bad_lines 为跳过未补写的坏行数。成功补写的行从
+    outbox 清除（文件重写为剩余行）；全部成功时 outbox
+    被清空（文件保留为空文件，不是删除——删除会丢“曾
+    有失败”的痕迹，空文件 + 台账行即为补齐证据）。
+    record_fn 缺省为 attempt_ledger.record_attempt（调用
+    方可注入假函数做单测）；record_fn 抛异常的行计入
+    failed 并保留。调用方负责提交（本函数不 commit，
+    与 record_attempt 的调用方提交约定一致）。
+    """
+    if record_fn is None:
+        from cloud.attempt_ledger import record_attempt as record_fn
+    records, bad = read_failures(path)
+    remaining_records = []
+    replayed = 0
+    failed = 0
+    for item in records:
+        try:
+            record_fn(conn, item["task_id"], item["outcome"],
+                      is_defect=False)
+            replayed += 1
+        except Exception:
+            failed += 1
+            remaining_records.append(item)
+    # 坏行无法解析出 task/outcome，不能补写也不能丢弃：
+    # 原样保留在 outbox 里（本函数读不到原文，故坏行数
+    # 只计入 remaining，不重写坏行文本——重写会丢原文）。
+    # 为避免重写丢坏行：仅当 bad == 0 时重写文件；有坏行
+    # 时不重写，只返回统计，由运维人工处理坏行后再重放。
+    remaining = failed + bad
+    if bad == 0:
+        try:
+            with open(path, "w", encoding="utf-8") as handle:
+                for item in remaining_records:
+                    handle.write(json.dumps(
+                        item, ensure_ascii=False) + "\n")
+        except OSError:
+            pass
+    return {
+        "replayed": replayed,
+        "failed": failed,
+        "remaining": remaining,
+        "bad_lines": bad,
+    }

cloud/quota.py

@@ -14,6 +14,18 @@ automations._automation_quota_exceeded），且三处已经
   绝不能触发上限；服务端 _quota_unpriced_tokens 改为本
   模块口径（行为变化仅限含 estimated 无价行的用户：此
   前可能被估算行误拦，此后不再误拦；真实花费的拦阻不变）。
+  集成 v2 确认（临时口径，非原 PR 已定）：本收敛来自
+  集成基底（非 main），原 PR 语义（main 里 #86 计全部
+  vs #90 actual-only 的分歧）并未在 main 上改动；合轨
+  时须由协调员另行确认 actual-only 是否成为原 PR 口径，
+  本轨只把集成 v2 的口径钉死并补双侧回归测试（见
+  cloud/tests/test_v2_evidence.py），绝不把临时口径写
+  成原 PR 已定。受影响调用方：server._quota_snapshot /
+  _check_unpriced_token_cap、worker._quota_limit_and_used /
+  _quota_exceeded、automations._automation_quota_exceeded，
+  三处均经本模块 monthly_usage + decide，改一处即三处
+  同口径；USD 求和不受影响（estimated 行 usd 为 NULL，
+  SUM 本就跳过）。
 - 默认配额解析：server、worker、automations 各一份
   CLOUD_DEFAULT_QUOTA_USD_MICROS 解析，现统一为
   default_quota_usd_micros（默认值 $10/月 = 10_000_000

cloud/server.py

@@ -96,6 +96,9 @@ from cloud.peer_messages import ensure_followups_table as _peer_ensure_followups
 from cloud.peer_messages import peer_message_prefix as _peer_prefix
 from cloud.peer_messages import row_get as _row_get
 from cloud.marker_tools import parse_spawn_defaults as _parse_spawn_defaults
+from cloud.evidence_binding import EvidenceNonceStore as _EvidenceNonceStore
+from cloud.evidence_binding import binding_presence as _evidence_presence
+from cloud.evidence_binding import validate_bound_evidence as _validate_evidence
 from cloud.merge_gate import evaluate_merge_gate as _merge_gate_evaluate
 from cloud.merge_gate import protection_from_env as _merge_gate_protection
 from cloud.audit_wiring import pushback_block_for_task as _pushback_block
@@ -6883,7 +6886,14 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au
         reader 强制；deploy.py 是运维脚本，不在本接口门禁内。
         本服务没有可信渠道直连 GitHub 核实 CI/保护：门禁强制的是
         “调用方提交的证据齐全且全绿”，谎报证据由审计与 PR/Checks
-        只读核对覆盖，不在本接口的验证范围内。
+        只读核对覆盖，不在本接口的验证范围内。集成 v2 补充
+        （cloud/gate_verification）：只读核对器本身已实现
+        （注入式 reader、逐项比对、mismatch 列出），但当前
+        合并/发布路径因请求体与任务行均无 repo/ref、且无可
+        信求证通道保证而接不上实时求证——见该模块头的不可
+        实现证明；当前真实控制仍是“证据齐全全绿 + 审计记
+        outcome/blocks”，审计 detail 不存证据体全文，谎报
+        发现靠事后比对，不是当场发现。
         """
         row = self._fetch_task(task_id)
         if row is None:
@@ -7444,7 +7454,16 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au
         把它绑到任务链上会在最需要部署修复与回滚时反锁部署。
         200 时写一条显式的 release_task 审计作为发布后动作记
         录（检查本身是只读的，这条是检查完成后的显式动作，
-        不是检查的副产品；锚点须在这条之后重取）。"""
+        不是检查的副产品；锚点须在这条之后重取）。
+        证据绑定（集成 v2，cloud/evidence_binding）：请求体成
+        对带 evidence_ts / evidence_nonce 时，服务端校验新鲜
+        度窗口并单次消费 nonce，原样重交即 409
+        release_blocked_evidence_replayed，陈旧/未来时间戳即
+        409 release_blocked_evidence_stale；两者皆无时走旧路径
+        并在成功响应标注 evidence_binding=absent。绑定防原样
+        重放与陈旧证据，不防重盖戳（无签名）；实时求证见
+        cloud/gate_verification（当前路径因无 repo/ref 且无
+        可信求证通道而接不上，见该模块头的不可实现证明）。"""
         row = self._fetch_task(task_id)
         if row is None:
             self._send_json(404, {"error": "task not found"})
@@ -7466,6 +7485,58 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au
         if hard_err:
             self._send_json(400, {"error": hard_err})
             return
+        # 证据新鲜度与单次使用绑定（集成 v2，cloud/evidence_binding）：
+        # 体里成对带 evidence_ts / evidence_nonce 时才校验；两者
+        # 皆无走旧路径（#259 契约兼容，成功响应标注 absent）；
+        # 只带其一是格式错误 400。校验通过的 nonce 在此次提交
+        # 即消费（无论后续门禁放行与否），原样重交同一 nonce
+        # 即 409 evidence_replayed；陈旧/未来时间戳即 409
+        # evidence_stale。409 按本接口规则不写审计；nonce
+        # 消费写 evidence_nonces 表（与审计链无关，不移动链头）。
+        # 绑定防原样重放与陈旧证据，不防重盖戳（无签名证据
+        # 无法识破旧内容盖新戳——见 evidence_binding 模块头）。
+        binding = _evidence_presence(body)
+        if binding == "partial":
+            self._send_json(400, {"error":
+                                  "evidence_ts and evidence_nonce"
+                                  " must be supplied together"})
+            return
+        if binding == "bound":
+            ok, reason = _validate_evidence(body)
+            if not ok:
+                if reason in ("evidence_too_old",
+                              "evidence_from_future"):
+                    self._send_json(409, {
+                        "error": "release blocked",
+                        "release_gate": {
+                            "allowed": False, "http_status": 409,
+                            "code": "release_blocked_evidence_stale",
+                            "blocked_by": "evidence_binding",
+                            "blocked_sources": ["evidence_binding"],
+                        },
+                    })
+                else:
+                    self._send_json(400, {"error": reason})
+                return
+            _bconn = self._db()
+            try:
+                _store = _EvidenceNonceStore(_bconn)
+                if not _store.consume(task_id,
+                                      body["evidence_nonce"],
+                                      body["evidence_ts"]):
+                    self._send_json(409, {
+                        "error": "release blocked",
+                        "release_gate": {
+                            "allowed": False, "http_status": 409,
+                            "code": "release_blocked_evidence_replayed",
+                            "blocked_by": "evidence_binding",
+                            "blocked_sources": ["evidence_binding"],
+                        },
+                    })
+                    return
+                _bconn.commit()
+            finally:
+                _bconn.close()
         # 发布时刻的新证据：体里没给任何硬门禁字段时，hard_gate
         # 为 None -> 证据缺失 409，而不是用合并时的旧结论补。
         has_evidence = any(
@@ -7512,6 +7583,10 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au
             store.audit(user_id, task["project_id"], "release_task",
                         task_id)
         task["release_gate"] = decision
+        # 绑定出示标注（绝不谎称）：bound = 本次提交带时间戳
+        # /nonce 且通过单次校验；absent = 未出示绑定，走 #259
+        # 旧路径（证据仍经硬门禁重算，但新鲜度未绑定）。
+        task["evidence_binding"] = binding
         self._send_json(200, task)
 
     # -- 运行时审批队列 ------------------------------------------------

cloud/tests/test_v2_evidence.py

@@ -0,0 +1,487 @@
+"""集成 v2 证据与口径硬伤定向测试（本轨新增，5 项全覆盖）。
+
+1. 证据重放绑定（cloud/evidence_binding + release 端点）：
+   原样重交被拒、正常证据通过、陈旧/未来被拒、只带其一
+   400、无绑定走旧路径并标注 absent；另钉住诚实边界：
+   绑定不鉴别内容（改 ci_status 后仍通过校验——防重盖
+   戳须签名或只读求证，不在本模块冒充）。
+2. 无价 token 口径统一（cloud/quota.monthly_usage 为
+   server/worker/automations 三处共用 canonical 源）：
+   actual 无价行计入、estimated 无价行不计入，并与
+   main 旧 SQL（计全部 NULL 行）做前后对比。
+3. 台账钩子开箱（cloud/ledger_onboarding）：表缺时
+   not ready、表齐且钩子开时 ready、表齐但钩子关时
+   仍 not ready；DDL 清单引用常量本身，不复制文本。
+4. 台账写失败 outbox（cloud/ledger_outbox + worker
+   _record_ledger_attempt 接线）：失败进 outbox、重放
+   补写行、重放后 outbox 清空、二次重放 0 行；任务
+   侧返回值仍为 None（不挡任务，规则 7 不变）。
+5. 只读核对器（cloud/gate_verification）：假 reader
+   下一致通过、谎报 CI/保护被 mismatch、无 reader /
+   无目标 / reader 失败一律 checked=False 且 verified
+   恒 False（skipped 绝不当 pass）；实时接线不可实现
+   的三点见 gate_verification 模块头，服务端 docstring
+   已同步该边界（本文件不测不存在的实时求证）。
+"""
+
+import http.client
+import json
+import logging
+import os
+import sqlite3
+import tempfile
+import threading
+import time
+import unittest
+from pathlib import Path
+
+try:
+    from cloud import ledger_onboarding
+    from cloud import ledger_outbox
+    from cloud.attempt_hook import ensure_marker_table
+    from cloud.attempt_ledger import PG_DDL as LEDGER_PG_DDL
+    from cloud.attempt_ledger import record_attempt
+    from cloud.attempt_hook import PG_MARKER_DDL as MARKER_PG_DDL
+    from cloud.cost import CostStore
+    from cloud.evidence_binding import EvidenceNonceStore
+    from cloud.evidence_binding import bind_evidence
+    from cloud.evidence_binding import binding_presence
+    from cloud.evidence_binding import validate_bound_evidence
+    from cloud.gate_verification import verify_gate_evidence
+    from cloud.quota import decide as quota_decide
+    from cloud.quota import monthly_usage
+except ImportError:
+    import sys
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud import ledger_onboarding
+    from cloud import ledger_outbox
+    from cloud.attempt_hook import ensure_marker_table
+    from cloud.attempt_ledger import PG_DDL as LEDGER_PG_DDL
+    from cloud.attempt_ledger import record_attempt
+    from cloud.attempt_hook import PG_MARKER_DDL as MARKER_PG_DDL
+    from cloud.cost import CostStore
+    from cloud.evidence_binding import EvidenceNonceStore
+    from cloud.evidence_binding import bind_evidence
+    from cloud.evidence_binding import binding_presence
+    from cloud.evidence_binding import validate_bound_evidence
+    from cloud.gate_verification import verify_gate_evidence
+    from cloud.quota import decide as quota_decide
+    from cloud.quota import monthly_usage
+
+
+def _hard_body():
+    return {
+        "files_changed": ["apps/web/a.py"],
+        "owned_dir": "apps/web",
+        "branch_protection": "protected",
+        "ci_status": "success",
+        "ci_checks": [{"name": "ci", "status": "success"}],
+        "required_checks": ["ci"],
+        "contract_version": "v2",
+        "contract_frozen": True,
+        "coordinator_approved": True,
+        "unresolved_review_comments": 0,
+    }
+
+
+class EvidenceBindingTestCase(unittest.TestCase):
+    def test_bind_and_validate_fresh(self):
+        stamped = bind_evidence(_hard_body())
+        self.assertEqual(binding_presence(stamped), "bound")
+        ok, reason = validate_bound_evidence(stamped)
+        self.assertTrue(ok)
+        self.assertIsNone(reason)
+
+    def test_partial_binding_is_not_bound(self):
+        body = _hard_body()
+        body["evidence_ts"] = time.time()
+        self.assertEqual(binding_presence(body), "partial")
+        ok, reason = validate_bound_evidence(body)
+        self.assertFalse(ok)
+        self.assertEqual(reason, "evidence_binding_incomplete")
+
+    def test_stale_and_future_rejected(self):
+        now = time.time()
+        old = bind_evidence(_hard_body(), now=now - 7200)
+        ok, reason = validate_bound_evidence(old, now=now)
+        self.assertFalse(ok)
+        self.assertEqual(reason, "evidence_too_old")
+        future = bind_evidence(_hard_body(), now=now + 3600)
+        ok, reason = validate_bound_evidence(future, now=now)
+        self.assertFalse(ok)
+        self.assertEqual(reason, "evidence_from_future")
+
+    def test_nonce_single_use_via_store(self):
+        conn = sqlite3.connect(":memory:")
+        store = EvidenceNonceStore(conn)
+        stamped = bind_evidence(_hard_body(), nonce="nonce-1")
+        self.assertTrue(store.consume("t1", stamped["evidence_nonce"],
+                                      stamped["evidence_ts"]))
+        conn.commit()
+        # 原样重交：同一 nonce 第二次消费即重放。
+        self.assertFalse(store.consume("t1", stamped["evidence_nonce"],
+                                       stamped["evidence_ts"]))
+
+    def test_binding_does_not_authenticate_content(self):
+        """诚实边界：绑定防原样重放，不防重盖戳/篡改内容。
+
+        盖戳后改 ci_status，校验仍通过——因为没有签名。
+        本项钉住该边界，防止后来人把 binding 误读成防篡改。
+        """
+        stamped = bind_evidence(_hard_body())
+        stamped["ci_status"] = "failure"
+        ok, _reason = validate_bound_evidence(stamped)
+        self.assertTrue(ok)
+
+
+class QuotaActualOnlyTestCase(unittest.TestCase):
+    def test_estimated_unpriced_rows_do_not_trip_cap(self):
+        tmp = tempfile.TemporaryDirectory()
+        try:
+            db_path = os.path.join(tmp.name, "control.db")
+            store = CostStore(db_path)
+            conn = sqlite3.connect(db_path)
+            try:
+                conn.execute("CREATE TABLE tasks(id TEXT PRIMARY KEY,"
+                             " created_by TEXT)")
+                conn.execute("INSERT INTO tasks(id, created_by)"
+                             " VALUES(?, ?)", ("task-1", "user-1"))
+                conn.commit()
+            finally:
+                conn.close()
+            # actual 无价 100（60+40），estimated 无价 5000，
+            # actual 有价 1 行（usd 非 NULL，不进无价计数）。
+            store.record("task-1", "proj", "terra-model",
+                         input_tokens=60, output_tokens=40,
+                         usd_micros=None, source="actual")
+            store.record("task-1", "proj", "terra-model",
+                         input_tokens=5000, output_tokens=0,
+                         usd_micros=None, source="estimated")
+            store.record("task-1", "proj", "gpt-priced",
+                         input_tokens=10, output_tokens=10,
+                         usd_micros=12345, source="actual")
+            month = time.strftime("%Y-%m", time.gmtime())
+            conn = sqlite3.connect(db_path)
+            try:
+                used, unpriced = monthly_usage(conn, "user-1", month)
+                # 前后对比：main 旧 SQL（计全部 NULL 行）会得
+                # 5100；actual-only 口径只得 100。
+                old_row = conn.execute(
+                    "SELECT COALESCE(SUM(c.input_tokens"
+                    " + c.output_tokens), 0) FROM cost_records c"
+                    " JOIN tasks t ON c.task_id = t.id"
+                    " WHERE t.created_by = ? AND c.day LIKE ?"
+                    " AND c.usd_micros IS NULL",
+                    ("user-1", month + "%")).fetchone()
+                self.assertEqual(int(old_row[0]), 5100)
+                self.assertEqual(unpriced, 100)
+                self.assertEqual(used, 12345)
+                # 双侧同一口径：CostStore 报表 unpriced 亦为 100。
+                report = store.report_by_task("task-1")
+                self.assertEqual(report["unpriced_actual_tokens"], 100)
+                # 上限判定：100 未超 1000 放行；5100（旧口径）
+                # 会误拦——这正是统一前服务端与执行侧的分歧点。
+                self.assertIsNone(quota_decide(10000000, used, unpriced,
+                                               1000, month))
+                self.assertIsNotNone(quota_decide(
+                    10000000, used, 5100, 1000, month))
+            finally:
+                conn.close()
+        finally:
+            tmp.cleanup()
+
+
+class LedgerOnboardingTestCase(unittest.TestCase):
+    def test_missing_tables_not_ready(self):
+        conn = sqlite3.connect(":memory:")
+        report = ledger_onboarding.ledger_setup_report(
+            conn, hook_enabled=True)
+        self.assertFalse(report["ready"])
+        self.assertFalse(report["tables_ready"])
+        self.assertIn("task_attempts table missing",
+                      report["missing"])
+        self.assertIn("task_attempt_origin_markers table missing",
+                      report["missing"])
+
+    def test_tables_ready_but_hook_off_not_ready(self):
+        conn = sqlite3.connect(":memory:")
+        record_attempt(conn, "task-1", "success", is_defect=False)
+        conn.commit()
+        ensure_marker_table(conn)
+        conn.commit()
+        report = ledger_onboarding.ledger_setup_report(
+            conn, hook_enabled=False)
+        self.assertTrue(report["tables_ready"])
+        self.assertFalse(report["ready"])
+        self.assertTrue(any("enable the hook" in s
+                            for s in report["next_steps"]))
+        report_on = ledger_onboarding.ledger_setup_report(
+            conn, hook_enabled=True)
+        self.assertTrue(report_on["ready"])
+
+    def test_ddl_checklist_references_constants_not_copies(self):
+        sources = ledger_onboarding.ledger_ddl_sources()
+        self.assertEqual(sources["task_attempts_pg_ddl"],
+                         LEDGER_PG_DDL)
+        self.assertEqual(sources["marker_pg_ddl"], MARKER_PG_DDL)
+        self.assertIn("task_attempts", LEDGER_PG_DDL)
+        self.assertIn("corrects_attempt_no", LEDGER_PG_DDL)
+        self.assertIn("task_attempt_origin_markers", MARKER_PG_DDL)
+        steps = [item["step"] for item in
+                 ledger_onboarding.LEDGER_DDL_CHECKLIST]
+        self.assertEqual(steps, [1, 2, 3, 4, 5])
+
+
+class LedgerOutboxTestCase(unittest.TestCase):
+    def test_append_and_replay(self):
+        tmp = tempfile.TemporaryDirectory()
+        try:
+            path = os.path.join(tmp.name, "failures.jsonl")
+            self.assertTrue(ledger_outbox.append_failure(
+                path, "task-1", "success", "boom"))
+            records, bad = ledger_outbox.read_failures(path)
+            self.assertEqual(len(records), 1)
+            self.assertEqual(bad, 0)
+            conn = sqlite3.connect(":memory:")
+            stats = ledger_outbox.replay_failures(path, conn)
+            conn.commit()
+            self.assertEqual(stats["replayed"], 1)
+            self.assertEqual(stats["remaining"], 0)
+            rows = conn.execute(
+                "SELECT outcome FROM task_attempts"
+                " WHERE task_id = ?", ("task-1",)).fetchall()
+            self.assertEqual([row[0] for row in rows], ["success"])
+            # 二次重放：outbox 已空，0 行，不会多写一行。
+            stats2 = ledger_outbox.replay_failures(path, conn)
+            self.assertEqual(stats2["replayed"], 0)
+            rows = conn.execute(
+                "SELECT COUNT(*) FROM task_attempts"
+                " WHERE task_id = ?", ("task-1",)).fetchone()
+            self.assertEqual(rows[0], 1)
+        finally:
+            tmp.cleanup()
+
+    def test_append_failure_never_raises(self):
+        # 路径父级是个文件时 makedirs 失败：返回 False，不抛。
+        tmp = tempfile.TemporaryDirectory()
+        try:
+            blocker = os.path.join(tmp.name, "blocker")
+            with open(blocker, "w") as handle:
+                handle.write("x")
+            bad_path = os.path.join(blocker, "failures.jsonl")
+            self.assertFalse(ledger_outbox.append_failure(
+                bad_path, "task-1", "success", "boom"))
+        finally:
+            tmp.cleanup()
+
+    def test_worker_record_failure_goes_to_outbox(self):
+        """worker 接线：_record_ledger_attempt 写失败时，
+        除日志外 outbox 里留一行，且返回 None（不挡任务）。"""
+        import cloud.worker as worker_mod
+        tmp = tempfile.TemporaryDirectory()
+        try:
+            db_path = os.path.join(tmp.name, "control.db")
+            worker = worker_mod.Worker.__new__(worker_mod.Worker)
+            worker.db_path = db_path
+            worker.attempt_ledger_hook = True
+            worker.logger = logging.getLogger("test-v2-evidence")
+            original_connect = worker_mod._db_connect
+
+            def _boom(_path):
+                raise RuntimeError("db down")
+
+            worker_mod._db_connect = _boom
+            try:
+                result = worker._record_ledger_attempt(
+                    "task-9", "success", True)
+            finally:
+                worker_mod._db_connect = original_connect
+            self.assertIsNone(result)
+            outbox = ledger_outbox.outbox_path_for(db_path)
+            records, bad = ledger_outbox.read_failures(outbox)
+            self.assertEqual(len(records), 1)
+            self.assertEqual(records[0]["task_id"], "task-9")
+            self.assertEqual(records[0]["outcome"], "success")
+            self.assertEqual(bad, 0)
+        finally:
+            tmp.cleanup()
+
+
+class GateVerificationTestCase(unittest.TestCase):
+    def _live(self, checks, protection):
+        def reader(repo, ref):
+            self.assertEqual(repo, "o/r")
+            self.assertEqual(ref, "sha1")
+            return {"ci_checks": checks,
+                    "branch_protection": protection}
+        return reader
+
+    def test_matching_evidence_verified(self):
+        result = verify_gate_evidence(
+            _hard_body(), reader=self._live(
+                {"ci": "success"}, "protected"),
+            repo="o/r", ref="sha1")
+        self.assertTrue(result["checked"])
+        self.assertTrue(result["verified"])
+        self.assertEqual(result["mismatches"], [])
+
+    def test_lied_ci_and_protection_mismatch(self):
+        result = verify_gate_evidence(
+            _hard_body(), reader=self._live(
+                {"ci": "failure"}, "unprotected"),
+            repo="o/r", ref="sha1")
+        self.assertTrue(result["checked"])
+        self.assertFalse(result["verified"])
+        self.assertTrue(any("ci_not_green_live" in m
+                            for m in result["mismatches"]))
+        self.assertTrue(any("branch_protection" in m
+                            for m in result["mismatches"]))
+
+    def test_no_reader_is_not_a_pass(self):
+        result = verify_gate_evidence(_hard_body(), reader=None,
+                                      repo="o/r", ref="sha1")
+        self.assertFalse(result["checked"])
+        self.assertFalse(result["verified"])
+        self.assertEqual(result["reason"], "no_trusted_reader")
+
+    def test_no_target_or_reader_failure_is_not_a_pass(self):
+        result = verify_gate_evidence(
+            _hard_body(), reader=self._live({}, "protected"),
+            repo=None, ref=None)
+        self.assertFalse(result["checked"])
+        self.assertFalse(result["verified"])
+
+        def _boom(_repo, _ref):
+            raise RuntimeError("github down")
+
+        result = verify_gate_evidence(_hard_body(), reader=_boom,
+                                      repo="o/r", ref="sha1")
+        self.assertFalse(result["checked"])
+        self.assertFalse(result["verified"])
+        self.assertTrue(result["reason"].startswith("reader_failed"))
+
+
+class ReleaseEvidenceBindingServerTestCase(unittest.TestCase):
+    def setUp(self):
+        from cloud.auth import AuthStore
+        from cloud.server import create_server
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self._saved_env = {
+            k: os.environ.get(k) for k in (
+                "CLOUD_MERGE_GATE_MODE", "CLOUD_BRANCH_PROTECTION",
+                "CODEX_GLOBAL_ADMINS")}
+        os.environ.pop("CLOUD_MERGE_GATE_MODE", None)
+        os.environ.pop("CLOUD_BRANCH_PROTECTION", None)
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id
+        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)
+        status, task = self._req(
+            "POST", "/tasks",
+            {"project_id": self.project["id"], "prompt": "x"},
+            self.owner_key)
+        self.assertEqual(status, 201)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "UPDATE tasks SET status = 'completed' WHERE id = ?",
+                (task["id"],))
+            conn.commit()
+        finally:
+            conn.close()
+        status, _merged = self._req(
+            "POST", "/tasks/%s/merge" % task["id"], _hard_body(),
+            self.owner_key)
+        self.assertEqual(status, 200)
+        self.task_id = task["id"]
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        for key, value in self._saved_env.items():
+            if value is None:
+                os.environ.pop(key, None)
+            else:
+                os.environ[key] = value
+        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 body is not None:
+            payload = json.dumps(body).encode("utf-8")
+            hdrs["Content-Type"] = "application/json"
+        if key is not None:
+            hdrs["X-API-Key"] = key
+        try:
+            conn.request(method, path, body=payload, 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
+
+    def test_replay_rejected_fresh_passes(self):
+        bound = bind_evidence(_hard_body())
+        status, first = self._req(
+            "POST", "/tasks/%s/release" % self.task_id, bound,
+            self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(first["evidence_binding"], "bound")
+        # 原样重交同一份证据体：nonce 已消费，409 重放。
+        status, replay = self._req(
+            "POST", "/tasks/%s/release" % self.task_id, bound,
+            self.owner_key)
+        self.assertEqual(status, 409)
+        self.assertEqual(replay["release_gate"]["code"],
+                         "release_blocked_evidence_replayed")
+        # 新 nonce 的正常证据仍通过（其余字段完全相同）。
+        fresh = bind_evidence(_hard_body())
+        status, second = self._req(
+            "POST", "/tasks/%s/release" % self.task_id, fresh,
+            self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(second["evidence_binding"], "bound")
+
+    def test_stale_partial_and_absent(self):
+        stale = bind_evidence(_hard_body(), now=time.time() - 7200)
+        status, resp = self._req(
+            "POST", "/tasks/%s/release" % self.task_id, stale,
+            self.owner_key)
+        self.assertEqual(status, 409)
+        self.assertEqual(resp["release_gate"]["code"],
+                         "release_blocked_evidence_stale")
+        partial = _hard_body()
+        partial["evidence_ts"] = time.time()
+        status, _resp = self._req(
+            "POST", "/tasks/%s/release" % self.task_id, partial,
+            self.owner_key)
+        self.assertEqual(status, 400)
+        status, plain = self._req(
+            "POST", "/tasks/%s/release" % self.task_id, _hard_body(),
+            self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(plain["evidence_binding"], "absent")
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/worker.py

@@ -7809,6 +7809,22 @@ class Worker:
         except Exception as exc:
             log_event(self.logger, "worker_attempt_ledger_failed",
                       message="attempt ledger write failed: %s" % exc)
+            # 集成 v2（cloud/ledger_outbox）：写失败除日志外再
+            # 进 outbox，运维事后可重放补写——任务状态与返回值
+            # 仍不受影响（规则 7 原样保留，台账不改 fail-closed，
+            # 审计 fail-closed 也不动；不对称的剩余部分见
+            # ledger_outbox 模块头的边界说明）。outbox 追加本身
+            # 也是 best-effort，失败只吞掉，不抛、不挡任务。
+            try:
+                from cloud.ledger_outbox import append_failure \
+                    as _ledger_outbox_append
+                from cloud.ledger_outbox import outbox_path_for \
+                    as _ledger_outbox_path
+                _ledger_outbox_append(
+                    _ledger_outbox_path(self.db_path),
+                    task_id, outcome, exc)
+            except Exception:
+                pass
             return None
 
     def run_one(self, task=None):