codex-app-server/cloud/ledger_outbox.py

"""台账写失败的 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,
    }