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/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,
+    }