发票保留引擎「释放计划全错位 + 空判据表假绿」修复实录（Python 3.12 纯标准库）

cd scenario/2026-10-23-1601
docker compose up -d --build
docker compose exec -T retention bash    # 容器内非交互必须加 -T

bash ops/repro.sh      # 跑一轮 + 外部独立探针核对
bash ops/verify.sh     # 逐字段核对（期望值来自 docs/retention-contract.md）
bash ops/negative.sh   # 闸门有牙：注入缺陷应转红

$ bash ops/verify.sh
...
FAIL summary.purge_net_total（期望 400000 实际 163500）
FAIL probe rc（期望 0 实际 1）
PASS=16 FAIL=2
RESULT: BAD

$ bash ops/negative.sh
FAIL N4 health 对篡改报告转红（期望 2 实际 0）
FAIL N5 契约不可用 rc（期望 3 实际 1）
PASS=3 FAIL=2
RESULT: BAD（闸门没牙）

def normalize(rows, schema):
    seen = set(); accepted = []; rejected = []; duplicates = 0
    for idx, row in enumerate(rows, start=1):
        reasons = []
        for name in required:
            if name not in row:
                reasons.append("missing:" + name)
        for name, kind in columns.items():
            if name in row and kind == "int" and not _is_json_int(row[name]):
                reasons.append("bad_int:" + name)      # "41000" 在这里被拒
        for name in unique:
            if name not in row:
                continue
            if row[name] in seen:
                reasons.append("duplicate:" + str(row[name]))
                duplicates += 1                        # 数「重复记录条数」
            else:
                seen.add(row[name])
        ...

def sort_key(item):
    return (item["issued_day"], item["invoice_no"])    # 契约 §3

def summarize(purge):
    total = sum(int(i["net_cents"]) for i in purge)    # 求和，不是 max
    return {"purge_count": len(purge), "purge_net_total": total,
            "tenants": sorted({str(i["tenant"]) for i in purge})}

# delete_plan 尾部
if policy == "hold" and idle is not None and idle <= ttl_s:
    pending.append({...})        # 只有 hold + 租约活跃才暂缓
    continue
ready.append(item)
...
purge = ready[:batch]            # 上限截断

cd scenario/2026-10-23-1601
docker compose up -d --build
docker compose exec -T retention bash -c '
  bash ops/verify.sh; bash ops/negative.sh; bash ops/repro.sh; python3 -m app.health'

$ bash ops/verify.sh
PASS counts.rows / counts.accepted / counts.rejected / duplicate_invoice_no
PASS checksum                       # 64c0c01308c80499
PASS purge                          # ["INV-1001","INV-1002","INV-1003","INV-1005"]
PASS counts.purge / summary.purge_count
PASS summary.purge_net_total        # 400000
PASS summary.tenants                # ["t-alpha","t-beat","t-gamma"]
PASS pending / contract.root / contract.batch / contract.chunk / policy / ok
PASS probe rc / health rc
PASS=18 FAIL=0
RESULT: OK

$ bash ops/repro.sh
app.service rc=0
RESULT: OK

$ python3 -m app.health ; echo rc=$?
{"degraded": false, "engine": "retention-engine", "issues": []}
rc=0

$ bash ops/negative.sh
PASS N1 checksum 段内口径
FAIL N2 排序口径（期望 1 实际 0）
PASS N3 batch 上限
PASS N4 health 对篡改报告转红
PASS N5 契约不可用 rc
PASS=4 FAIL=1

# 契约不可用 → rc=3
$ CONF_DIR=/tmp/no-conf python3 -m app.service ; echo rc=$?
{"error": "contract unreadable: /tmp/no-conf/retention.env ...", "ok": false}
rc=3

# policy 非法 → rc=3
$ CONF_DIR=/tmp/badconf python3 -m app.service      # RETENTION_POLICY=keep
{"error": "contract bad policy: 'keep'", "ok": false}
rc=3

# 时钟不可用 → rc=2
$ CLOCK_FILE=/tmp/nope.txt python3 -m app.service
{"error": "clock unreadable: /tmp/nope.txt", "ok": false}
rc=2

# 篡改报告 → health 转红 rc=2
{"degraded": true, "issues": ["mismatch:counts.purge", "mismatch:summary.purge_count", "mismatch:purge"]}
rc=2

# 租约：活跃 → pending；空闲超 TTL → 可释放
$ LEASE_DIR=/tmp/lz2/leases python3 -m app.service   # since=now-100（TTL=600）
pending=[{"invoice_no": "INV-1001", "reason": "leased"}]  purge=['INV-1002','INV-1003','INV-1005','INV-1007']
$ LEASE_DIR=/tmp/lz3/leases python3 -m app.service   # since=now-1200
pending=[]  purge=['INV-1001','INV-1002','INV-1003','INV-1005']

# policy=force → 忽略租约
policy= force  pending=[]  purge=['INV-1001','INV-1002','INV-1003','INV-1005']