codex-app-server/cloud/ledger_onboarding.py

"""台账钩子开箱闭合方案（集成 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,
    }