MQTT 3.1.1 Broker 会话投递审计：四盏绿灯账全错，健康检查自证健康怎么修

python3 -m app.service    # rc=0
python3 -m app.health     # {"degraded": false, "issues": []}
python3 -m app.run        # 不报错，reports/audit.json 字段齐全

cd scenario/2026-10-26-0002
docker compose up -d --build
docker compose exec -T broker bash      # 非交互必须加 -T

# 容器内（/srv）
bash ops/repro.sh   ; echo "probe_rc=$?"    # 问题态：RESULT: BAD（不一致 14 条）/ rc=1
bash ops/verify.sh  ; echo "verify_rc=$?"   # 问题态：PASS=21 FAIL=18 / RESULT: BAD
bash ops/negative.sh                        # 问题态：PASS=14 FAIL=2
python3 -m app.service ; echo "rc=$?"       # 问题态：rc=0（假的绿）

cd scenario/2026-10-26-0002
bash ops/repro.sh   ; echo rc=$?    # RESULT: BAD，rc=1（rc=1 是预期信号，用 ; 接住）
bash ops/verify.sh  ; echo rc=$?    # PASS=21 FAIL=18 / RESULT: BAD
bash ops/negative.sh                # PASS=14 FAIL=2（N14/N15 红）

# 模拟镜像 /srv 布局，确认 COPY 进容器的代码同样正确
rm -rf /tmp/srv && mkdir -p /tmp/srv/reports
cp -r app conf input docs client ops /tmp/srv/ && cd /tmp/srv
bash ops/repro.sh ; echo rc=$?      # RESULT: OK，rc=0
bash ops/verify.sh                  # PASS=39 FAIL=0 / RESULT: OK
bash ops/negative.sh                # PASS=16 FAIL=0 / RESULT: OK

报告算错  →  健康检查复用 audit 的计算  →  照错的再算一遍"一致"  →  报绿
        ↘  判据表是空元组，issues 恒 []  ↗
        ↘  return 0 写死，退出码分层形同虚设 ↗

class ContractError(Exception): ...

REQUIRED = {
    "now": lambda v: isinstance(v, int) and v >= 0,
    "session_expiry_seconds": lambda v: isinstance(v, int) and v > 0,
    "max_qos": lambda v: isinstance(v, int) and 0 <= v <= 2,
    "retain_handling": lambda v: isinstance(v, int) and v in (0, 1),
    "engine": lambda v: isinstance(v, str) and v != "",
}

def load(path=CONTRACT_PATH):
    try:
        with open(path, "r", encoding="utf-8") as f:
            conf = json.load(f)
    except OSError as e:
        raise ContractError("契约不可用: %s" % e)   # 不再静默回退默认值
    except ValueError as e:
        raise ContractError("契约不是合法 JSON: %s" % e)
    return validate(conf)

# 会话：过期只看业务口径，persisted 必须是 connected 且未过期
expired = (now - c["last_seen"]) > conf["session_expiry_seconds"]
persisted = c["connected"] and not expired
queued_count = 0 if (expired or c["connected"]) else 1

# 订阅：被拒订阅在加载阶段就丢弃，后面处处少一个 if
if loaders.as_int(r["granted_qos"]) < 0: continue

# 受众：dict + max 天然完成「同客户端去重 + 只抬 QoS」
hit[cid] = max(hit.get(cid, -1), row["granted_qos"])

# 保留：先取每主题 pub_seq 最大的 retain=1，再滤掉 payload_len=0（清除）
latest = {t: p for t, p in latest.items() if p["payload_len"] > 0}

# 有效 QoS：夹 max_qos，过期会话强制 0
sess["effective_qos"] = 0 if sess["expired"] else min(max(granted), cap)

# 传输：半帧不计字节，未投递只看末次尝试
sent_bytes = sum(t["sent_bytes"] for t in trans if t["frame_complete"])
undelivered = sum(1 for rows in groups.values()
                  if not sorted(rows, key=lambda x: x["attempt"])[-1]["ack_ok"])

def recompute(conf):
    """在 health 内部重新读 input/*.tsv 重算，不 import audit 的任何推导。"""
    ...

def compare(expected, report):
    """逐字段比对 totals/deliveries/transport/sessions[*]/per_topic[*]/ok，条数与值都查。"""
    ...

CHECKS = 11   # 非空：7 个统计字段 + 4 个语义字段

cd scenario/2026-10-26-0002
bash ops/verify.sh   ; echo rc=$?    # PASS=39 FAIL=0 TOTAL=39 / RESULT: OK
bash ops/negative.sh ; echo rc=$?    # PASS=16 FAIL=0 TOTAL=16 / RESULT: OK
bash ops/repro.sh    ; echo rc=$?    # 探针 RESULT: OK / rc=0

# 篡改自述字段 ok
python3 -c "import json;p='reports/audit.json';d=json.load(open(p));d['ok']=not d['ok'];json.dump(d,open(p,'w'))"
python3 -m app.health ; echo rc=$?     # rc=2（不再自证健康）

# 篡改统计值
python3 -c "import json;p='reports/audit.json';d=json.load(open(p));d['transport']['sent_bytes']=61;json.dump(d,open(p,'w'))"
python3 -m app.health ; echo rc=$?
# {"degraded": true, "issues": ["transport.sent_bytes: 期望 52 实际 61"], "rc": 2}