#!/usr/bin/env bash
# 15-criterion4-live.sh —— 契约 §6 **判据④** 的活环境验收门
#
#   判据④ 原文：≥2 台物理机的 Agent **在同一房间互见**，且各自被 @ 时命中。
#
# 为什么值得单独一个脚本：判据④ 是**阶段 1 的唯一硬阻塞**，而它此前只能肉眼看房间 -
# 谁在跑、谁被命中、有没有串台，全靠人记。本脚本把它变成"跑一次、看一张表、退出码说话"。
#
# 用法（凭据只走环境变量，不落盘、不打印）：
#   MATRIX_BASE_URL=https://imbot.apescale.com \
#   MATRIX_ACCESS_TOKEN=<本节点 bot token> \
#   BRIDGE_ROOM='!xxxx:matrix.113.249.102.nip.io' \
#   TARGET_A='@tutti-hermes:matrix.113.249.102.nip.io' \
#   TARGET_B='@tutti-pi:matrix.113.249.102.nip.io' \
#   BRIDGE_A='@bridge3:...' BRIDGE_B='@bridge2:...' \
#   NODE_NAME=mac13 \
#   bash scripts/15-criterion4-live.sh
#
#   bash scripts/15-criterion4-live.sh --selftest     # 离线自测（含"门必须会红"的对照）
#   bash scripts/15-criterion4-live.sh --pair a.log b.log   # 两台机各跑一次后, 机器判定"两台物理机"
#
# 退出码：0 = 判据④ 通过 ｜ 1 = 断言失败（门会红）｜ 2 = 前置缺失/取数不可信（拒绝判读）
# 边界：只向 BRIDGE_ROOM 白名单房间发消息（R-8）；不发任何其它房间；不打印任何凭据值。
set -uo pipefail

BASE="${MATRIX_BASE_URL:-}"
TOKEN="${MATRIX_ACCESS_TOKEN:-}"
ROOM="${BRIDGE_ROOM:-}"
TA="${TARGET_A:-}"
TB="${TARGET_B:-}"
BA="${BRIDGE_A:-}"
BB="${BRIDGE_B:-}"
NODE="${NODE_NAME:-$(hostname -s 2>/dev/null || echo unknown)}"
WAIT="${P4_WAIT:-45}"        # 等回执链的总超时(秒)
NEG_WAIT="${P4_NEG_WAIT:-8}" # 阴性对照观察窗(秒)
pass=0; fail=0; skip=0
ok()   { printf '  \033[32m✅\033[0m %s\n' "$*"; pass=$((pass+1)); }
bad()  { printf '  \033[31m❌\033[0m %s\n' "$*"; fail=$((fail+1)); }
skp()  { printf '  \033[33m⏭\033[0m  %s\n' "$*"; skip=$((skip+1)); }
sec()  { printf '\n\033[1m── %s\033[0m\n' "$*"; }

TMP="$(mktemp -d)"; trap 'rm -rf "$TMP"' EXIT
FETCH_FAIL=0      # 取数失败 ⇒ 拒绝判读(exit 2), 不与"断言失败(exit 1)"混为一谈

enc() { python3 -c "import urllib.parse,sys;print(urllib.parse.quote(sys.argv[1],safe=''))" "$1"; }

api() { # api <METHOD> <path> [json-file]  -> stdout; 返回 curl 退出码
  local m="$1" p="$2" f="${3:-}"
  if [ -n "$f" ]; then
    curl -sS --max-time 20 -X "$m" -H "Authorization: Bearer $TOKEN" \
      -H 'Content-Type: application/json' --data-binary "@$f" "$BASE$p"
  else
    curl -sS --max-time 20 -X "$m" -H "Authorization: Bearer $TOKEN" "$BASE$p"
  fi
}

need() { [ -n "${1:-}" ] || { echo "🔴 缺少环境变量 $2 —— 判读不可信，拒绝判读(exit 2)" >&2; return 2; }; }

preflight() {
  sec "前置（只读）"
  need "$BASE" MATRIX_BASE_URL; need "$TOKEN" MATRIX_ACCESS_TOKEN
  need "$ROOM" BRIDGE_ROOM;     need "$TA" TARGET_A; need "$TB" TARGET_B
  local r
  r="$(api GET /_matrix/client/v3/account/whoami)" || { bad "whoami 请求失败(curl)"; return 2; }
  ME="$(printf '%s' "$r" | python3 -c "
import json,sys
d=json.load(sys.stdin)
print(d.get('user_id','') if isinstance(d,dict) else '')")"
  [ -n "$ME" ] && ok "桥身份可认证: $ME" || { bad "whoami 拿不到 user_id(令牌失效?) —— 判读不可信"; return 2; }

  r="$(api GET /_matrix/client/v3/joined_rooms)"
  printf '%s' "$r" | python3 -c "
import json,sys
d=json.load(sys.stdin)
sys.exit(0 if isinstance(d,dict) and sys.argv[1] in (d.get('joined_rooms') or []) else 1)" "$ROOM" \
    && ok "已加入白名单房间 $ROOM" || { bad "该 token 未加入 ${ROOM}（或 joined_rooms 取数失败）"; return 2; }

  r="$(api GET "/_matrix/client/v3/rooms/$(enc "$ROOM")/joined_members")"
  MISS="$(printf '%s' "$r" | python3 -c "
import json,sys
d=json.load(sys.stdin)
if not isinstance(d,dict) or 'joined' not in d:
    print('APIERR'); sys.exit(0)
j=set(d['joined'])
print(','.join(t for t in sys.argv[1:3] if t not in j))" "$TA" "$TB")"
  case "$MISS" in
    APIERR) bad "joined_members 取数失败 —— 判读不可信"; return 2;;
    "")     ok "两个 target 都在房间里(R-7 前置): $(basename "${TA%%:*}") / $(basename "${TB%%:*}")";;
    *)      bad "房间缺 target: $MISS —— 判据④ 前置不满足(先做 T-1.10 步骤 1)"; return 2;;
  esac
  return 0
}

send() { # send <text> -> 打印 event_id
  local text="$1" f="$TMP/send.json" txn="p4-$$-$RANDOM"
  python3 -c "
import json,sys
json.dump({'msgtype':'m.text','body':sys.argv[1]}, open(sys.argv[2],'w',encoding='utf-8'), ensure_ascii=False)" \
    "$text" "$f"
  api PUT "/_matrix/client/v3/rooms/$(enc "$ROOM")/send/m.room.message/$(enc "$txn")" "$f" \
    | python3 -c "import json,sys;d=json.load(sys.stdin);print(d.get('event_id','') if isinstance(d,dict) else '')"
}

receipts() { # receipts <根事件id> -> 每行 "<sender> <status>"; 取数失败输出 __FETCH_FAILED__ 哨兵
  local root="$1" f="$TMP/msgs.json"
  api GET "/_matrix/client/v3/rooms/$(enc "$ROOM")/messages?dir=b&limit=100" > "$f"
  python3 - "$root" "$f" <<'PY'
import json, sys
root, path = sys.argv[1], sys.argv[2]
# 缺陷L(workbuddy-mac13 报): 取数失败**不许**与"确实 0 条"同形 —— 阴性对照的判据就是"0 条",
# 吞掉错误会让它 fail-open(403/401/网络故障都会判绿)。故用哨兵, 由消费处拒绝判读。
try:
    d = json.load(open(path, encoding="utf-8"))
except Exception:
    print("__FETCH_FAILED__"); sys.exit(0)
if not isinstance(d, dict) or "chunk" not in d:
    print("__FETCH_FAILED__"); sys.exit(0)
rows = []
for ev in (d.get('chunk') or []):
    c = ev.get('content') or {}
    rel = c.get('m.relates_to') or {}
    if rel.get('rel_type') != 'm.thread' or rel.get('event_id') != root:
        continue
    st = c.get('status') or c.get('tutti_status')
    if st:
        rows.append((ev.get('origin_server_ts', 0), ev.get('sender', '?'), st))
for _, sender, st in sorted(rows):          # 按服务端时间戳正序, 不信服务端返回顺序
    print(f"{sender} {st}")
PY
}

expect_chain() { # expect_chain <标签> <根事件id> <期望 sender(可空)> <禁止 sender(可空)>
  local label="$1" root="$2" want="$3" forbid="$4" got=""
  local deadline=$(( $(date +%s) + WAIT ))
  while :; do
    got="$(receipts "$root")"
    if printf '%s' "$got" | grep -q '__FETCH_FAILED__'; then
      bad "$label 取数失败(/messages) —— 判读不可信, 拒绝给结论"; FETCH_FAIL=1; return 2
    fi
    printf '%s' "$got" | grep -qE ' (succeeded|failed|timeout)$' && break
    [ "$(date +%s)" -ge "$deadline" ] && break
    sleep 2
  done
  local states senders
  states="$(printf '%s' "$got" | awk 'NF{print $2}' | paste -sd' ' - 2>/dev/null)"
  senders="$(printf '%s' "$got" | awk 'NF{print $1}' | sort -u | paste -sd' ' - 2>/dev/null)"
  printf '        %s 回执: [%s] 来自: [%s]\n' "$label" "${states:-无}" "${senders:-无}"
  # 1) 必须是 accepted -> running -> 终态, 且 running 不许缺(R-3)
  printf '%s' "$states" | grep -q "accepted" || { bad "$label 缺 accepted"; return 1; }
  printf '%s' "$states" | grep -q "running"  || { bad "$label 缺 running(R-3: 禁止静默完成)"; return 1; }
  printf '%s' "$states" | grep -qE "succeeded|failed|timeout" || { bad "$label 未达终态(超时 ${WAIT}s)"; return 1; }
  python3 -c "
import sys
seq=sys.argv[1].split()
# 【dsh@mac13】R-1 与 R-3/R-4 的**表面冲突**：契约 R-1 明写"重复投递返回上一次的回执、attempt 递增"，
#   于是合法序列可能是 `accepted accepted running succeeded`。老断言要求 accepted 只能在第 0 位，
#   会把这种**合法重放**误判成顺序错。改判为：首条必须是 accepted；accepted 不得晚于 running；
#   running 至多一条；终态后不得再有回执。（真链路实测出现过 attempt 1→2 两条 accepted。）
first_run = next((i for i, s in enumerate(seq) if s == 'running'), None)
first_acc = next((i for i, s in enumerate(seq) if s == 'accepted'), None)
bad = []
if first_acc != 0: bad.append('首条不是 accepted')
if any(s == 'accepted' and first_run is not None and i > first_run for i, s in enumerate(seq)):
    bad.append('accepted 出现在 running 之后')
if seq.count('running') > 1: bad.append('running 出现多次')
term=[i for i,s in enumerate(seq) if s in ('succeeded','failed','timeout')]
assert not bad, 'accepted/running 顺序不合 R-3/R-4: '+'; '.join(bad)+' | '+' '.join(seq)
assert not term or term[0]==len(seq)-1, '终态后仍有回执: '+sys.argv[1]" "$states" \
    || { bad "$label 回执顺序不合 R-3/R-4"; return 1; }
  ok "$label 回执链完整(到达终态, R-3): $states"
  # 【dsh@mac13】补第二档判定：上面只保证"出现了终态"(R-3 不许静默完成)，
  #   但 **A2 的真实语义是"结果回到房间"** ⇒ 失败/超时不该与成功同等计为通过。
  #   实测依据：接真 agent 后出现过 `[accepted running failed]` 被旧门判 ✅。
  if printf '%s' "$states" | grep -qE " (failed|timeout)$"; then
    bad "$label 终态是失败/超时(A2 未达成): $states"
    return 1
  fi
  ok "$label 终态为成功(A2 结果已回到房间)"
  # 2) 归属: 只应由服务该 target 的桥回执(R-7 一 target 一账号, 不许串台)
  if [ -n "$want" ]; then
    [ "$senders" = "$want" ] && ok "$label 回执归属正确: $want" \
      || { bad "$label 期望来自 $want, 实际 [$senders]"; return 1; }
  fi
  if [ -n "$forbid" ]; then
    case " $senders " in
      *" $forbid "*) bad "$label 出现**另一节点**的桥($forbid)回执 ⇒ 串台(R-7 违反)"; return 1;;
      *) ok "$label 未串台(无 $forbid 回执)";;
    esac
  fi
  return 0
}

probe() { # probe <label> <target> <期望sender> <禁止sender>
  local label="$1" target="$2" want="$3" forbid="$4" eid
  # 【dsh@mac13】探针正文可配置（**默认行为不变**）：实测 hermes CLI 对"短促+编号+方括号标签"
  #   这类文本会长时间不退出（对照：自然语言 ~40-100s 正常返回），导致 A2 只能拿到 failed。
  #   节点归属信息不丢 —— 两台机比对用的是下面的 P4-FINGERPRINT 行，不是消息尾巴。
  eid="$(send "$target ${P4_LABEL_PREFIX:-判据④ 探针} $(date +%H%M%S)${P4_PROBE_SUFFIX- [$NODE]}")"
  [ -n "$eid" ] || { bad "$label 发送失败(无 event_id)"; return 1; }
  printf '        根事件: %s\n' "$eid"
  ROOTS="$ROOTS $label=$eid"
  if expect_chain "$label" "$eid" "$want" "$forbid"; then
    # 机器可读行: 供 --pair 做"两台物理机"的**自动**判定, 不再靠人比对两行指纹。
    printf '        P4-HIT node=%s target=%s bridge=%s ok=1\n' "$NODE" "$target" "${want:-?}"
    return 0
  fi
  printf '        P4-HIT node=%s target=%s bridge=%s ok=0\n' "$NODE" "$target" "${want:-?}"
  return 1
}

negative_control() {
  local eid got n
  eid="$(send "判据④ 阴性对照:本条没有 @ 任何 target${P4_PROBE_SUFFIX- [$NODE]}")"
  [ -n "$eid" ] || { bad "阴性对照发送失败"; return 1; }
  sleep "$NEG_WAIT"
  got="$(receipts "$eid")"
  if printf '%s' "$got" | grep -q '__FETCH_FAILED__'; then
    bad "阴性对照取数失败(/messages) ⇒ 不能拿\"0 条\"当通过(缺陷L)"; FETCH_FAIL=1; return 2
  fi
  n="$(printf '%s' "$got" | grep -c . )"
  if [ "$n" -eq 0 ]; then ok "阴性对照: 无 @ 的消息在 ${NEG_WAIT}s 内 0 条回执(判据③ 活环境侧)";
  else bad "阴性对照失败: 无 @ 却收到 $n 条回执 ⇒ 误触发"; printf '%s\n' "$got" | sed 's/^/        /'; return 1; fi
}

pair_check() {   # pair_check <fileA> <fileB>: 两台机各跑一次的原始输出
  local a="$1" b="$2"
  for f in "$a" "$b"; do
    [ -f "$f" ] || { echo "🔴 找不到文件: $f" >&2; exit 2; }
  done
  python3 - "$a" "$b" <<'PY'
import re, sys
HIT = re.compile(r"P4-HIT node=(\S+) target=(\S+) bridge=(\S+) ok=(\d)")
FP = re.compile(r"P4-FINGERPRINT node=(\S+) user=(\S+) room=(\S+)")
def parse(path):
    hits, node, user, room = [], None, None, None
    for ln in open(path, encoding="utf-8", errors="replace"):
        m = HIT.search(ln)
        if m:
            hits.append(m.groups())
        f = FP.search(ln)
        if f:
            node, user, room = f.group(1), f.group(2), f.group(3)
    return hits, node, user, room
ha, na, ua, ra = parse(sys.argv[1])
hb, nb, ub, rb = parse(sys.argv[2])
fail, ok = [], []
if not ha or not hb:
    print("🔴 至少一侧没有 P4-HIT 行(该侧未跑或未通过) ⇒ 判读不可信"); sys.exit(2)
if not (na and ua and ra and nb and ub and rb):
    print("🔴 至少一侧缺指纹行(node/user/room) ⇒ 判读不可信"); sys.exit(2)

if na == nb:
    fail.append(f"两侧 node 相同({na}) ⇒ **不是两台物理机**")
else:
    ok.append(f"两个节点不同: {na} / {nb}")
if ua == ub:
    fail.append(f"两侧自称的桥身份相同({ua}) ⇒ 无法证明是两个桥进程")
else:
    ok.append(f"两个桥身份不同: {ua} / {ub}")
if ra != rb:
    fail.append(f"两侧房间不同: {ra} vs {rb} ⇒ 不在同一房间, 判据④ 不成立")
else:
    ok.append(f"同一房间: {ra}")

# 每轮: 所有探针必须 ok, 且每个 target 在**该轮内**只被一个桥应答(R-7)
maps, bad_probe = [], False
for tag, hits in (("A", ha), ("B", hb)):
    if [h for h in hits if h[3] != "1"]:
        fail.append(f"第 {tag} 轮存在未通过的探针: "
                    + ",".join(f"{h[0]}:{h[1]}" for h in hits if h[3] != "1"))
        bad_probe = True
    per = {}
    for _node, target, bridge, _ok in hits:
        per.setdefault(target, set()).add(bridge)
    for target, brs in sorted(per.items()):
        if len(brs) > 1:
            fail.append(f"第 {tag} 轮 target {target} 被多个桥应答 {sorted(brs)} ⇒ R-7 违反")
    maps.append(per)

# 缺陷K(workbuddy-mac13 报): 原判定是**条件式**的 —— 两侧都没有某 target 时两支都不进,
# 静默跳过 ⇒ 零交叉也判通过(阶段门放水)。现在改为**无条件**:
sa, sb = set(maps[0]), set(maps[1])
if sa != sb:
    fail.append(f"两侧 target 覆盖不一致: A={sorted(sa)} B={sorted(sb)} ⇒ "
                f"零交叉/单侧覆盖都**没验过**'各自被 @ 时命中'")
elif len(sa) < 2:
    fail.append(f"每侧只覆盖 {len(sa)} 个 target({sorted(sa)}) ⇒ 判据④ 要求两个 target 各自被 @ 命中")
else:
    ok.append(f"两侧覆盖一致且 ≥2 个 target: {sorted(sa)}")
for tg in sorted(sa & sb):
    a_br, b_br = sorted(maps[0][tg]), sorted(maps[1][tg])
    if a_br != b_br:
        fail.append(f"跨轮不一致: target {tg} 在 A 轮由 {a_br} 应答、B 轮由 {b_br} ⇒ 归属不稳定(R-7)")
    else:
        ok.append(f"跨轮一致: {tg} → {a_br[0]}")
bridges = {h[2] for h in (ha + hb)}
if len(bridges) < 2:
    fail.append(f"两次运行只见 1 个桥身份({sorted(bridges)}) ⇒ 无法证明两个节点各自独立应答")
else:
    ok.append(f"两个桥身份在场: {'/'.join(sorted(bridges))}")

# 绿勾只在**没有 fail** 时打(原写法 `not fail or all(...)` 会在 fail 非空时照打绿勾)
if not fail:
    print(f"  ✅ 每一轮内每个 target 唯一应答; 探针全部通过")
for x in ok:
    print(f"  ✅ {x}")
for x in fail:
    print(f"  ❌ {x}")
print()
if fail:
    print(f"❌ 判据④(两台物理机)未通过: {len(fail)} 项"); sys.exit(1)
print("✅ 判据④(两台物理机)通过: 两节点 · 两桥身份 · 同房间 · 每 target 归属唯一且跨轮一致")
PY
}

run_live() {
  local rc=0
  # 每次进来重取环境: 这样 --selftest 里 export 的替身配置才生效(脚本顶部的初值只在直接调用时有用)。
  BASE="${MATRIX_BASE_URL:-}"; TOKEN="${MATRIX_ACCESS_TOKEN:-}"; ROOM="${BRIDGE_ROOM:-}"
  TA="${TARGET_A:-}"; TB="${TARGET_B:-}"; BA="${BRIDGE_A:-}"; BB="${BRIDGE_B:-}"
  NODE="${NODE_NAME:-$(hostname -s 2>/dev/null || echo unknown)}"
  WAIT="${P4_WAIT:-45}"; NEG_WAIT="${P4_NEG_WAIT:-8}"
  ROOTS=""
  preflight || return 2
  sec "探针（真实发消息到白名单房间）"
  probe "判据④-A @ ${TA%%:*}" "$TA" "$BA" "${BB:-}" || rc=1
  probe "判据④-B @ ${TB%%:*}" "$TB" "$BB" "${BA:-}" || rc=1
  sec "阴性对照"
  negative_control || rc=1
  idempotency_control "$TA" "${BA:-$TA}" || rc=1
  sec "节点指纹（两台机各跑一次, 两行对比即为判据④ 的「两台物理机」证据）"
  printf '        P4-FINGERPRINT node=%s user=%s room=%s roots=%s\n' "$NODE" "${ME:-?}" "$ROOM" "${ROOTS# }"
  printf '\n\033[1m═══ 判据④: %d 通过 / %d 失败 / %d 跳过 ═══\033[0m\n' "$pass" "$fail" "$skip"
  if [ "$FETCH_FAIL" = "1" ]; then
    echo "🔴 判据④  **拒绝判读**: 有取数失败(见上) —— 不要把'取不到'当成'没发生'"; return 2
  fi
  [ "$rc" -eq 0 ] && { echo "✅ 判据④ 在本节点通过"; return 0; } || { echo "❌ 判据④ 未通过"; return 1; }
}


# ── 幂等复验（R-1/A4 的活环境形态）──────────────────────────────────
# ⚠️ 「重放」的正确定义（我上一版在这里想错了，留字免得后人照抄）：
#   契约 F2 是 idem_key = sha256(room|sender|client_ts|prompt_digest) —— **含时间戳**；
#   §6 判据 2 说的是「同一消息重放 3 次（同 idem_key）→ 只执行一次」。
#   ⇒ 连发三条正文相同的消息**不是重放**：各自 origin_server_ts 不同 ⇒ 本就该各执行一次
#     （实测 rig 里就是 3 个 task_id，全 done）。真正的重放是**同一条事件被重复投递**
#     （Matrix 重投、或桥重启后 initial sync 把旧事件再带回一次）。
#   所以这里：发一条 → 等它到终态 → 用 REPLAY_CMD 让同一条事件被重新投递 →
#   要求房间里**仍只有一个 task_id**，且重投确实带来了新回执（否则不许算通过）。
idempotency_control() { # idempotency_control <target> <期望应答者>
  local target="$1" want="$2"
  [ -n "$target" ] || { skip "幂等复验: 未给 TARGET_A ⇒ 跳过"; return 0; }
  if [ -z "${REPLAY_CMD:-}" ]; then
    skip "幂等复验: 未提供 REPLAY_CMD(重投同一条事件的手段) ⇒ 跳过，不假装通过"
    return 0
  fi
  sec "幂等复验（R-1/A4：同一条事件被重投）"
  local body eid roots="" all n_recv n_tasks n_recv2 wait_left t0
  wait_left="${IDEM_WAIT:-150}"
  body="$target ${P4_LABEL_PREFIX:-判据④ 探针} replay$(date +%H%M%S)${P4_PROBE_SUFFIX- [$NODE]}"
  eid="$(send "$body")"
  [ -n "$eid" ] || { bad "幂等复验发送失败"; return 1; }
  roots="$eid"
  printf '        原始事件: %s\n' "$eid"

  t0=$(date +%s)
  while :; do
    sleep 5
    all="$(probe_receipts "$roots")" || { bad "取数失败 ⇒ 拒绝判读(承缺陷L)"; FETCH_FAIL=1; return 2; }
    [[ "$all" == __FETCH_FAILED__* ]] && { bad "取数失败 ⇒ 拒绝判读(承缺陷L)"; FETCH_FAIL=1; return 2; }
    printf '%s\n' "$all" | grep -qE " (succeeded|failed|timeout)$" && break
    [ $(( $(date +%s) - t0 )) -ge "$wait_left" ] && break
  done
  n_recv="$(printf '%s\n' "$all" | grep -c .)"
  if [ "${n_recv:-0}" -eq 0 ]; then
    bad "第一轮就没收到任何回执 ⇒ 无法判定(不许把'没测到'当'通过')"; return 1
  fi
  ok "第一轮有回执 ${n_recv} 条，进入重投"

  say "   执行重投: $REPLAY_CMD"
  sh -c "$REPLAY_CMD" >/dev/null 2>&1 || { bad "重投命令失败 ⇒ 无法判定"; return 1; }
  t0=$(date +%s)
  while :; do
    sleep 5
    all="$(probe_receipts "$roots")" || { bad "取数失败 ⇒ 拒绝判读"; FETCH_FAIL=1; return 2; }
    [[ "$all" == __FETCH_FAILED__* ]] && { bad "取数失败 ⇒ 拒绝判读"; FETCH_FAIL=1; return 2; }
    n_recv2="$(printf '%s\n' "$all" | grep -c .)"
    [ "${n_recv2:-0}" -gt "${n_recv:-0}" ] && break
    [ $(( $(date +%s) - t0 )) -ge "$wait_left" ] && break
  done
  n_recv2="$(printf '%s\n' "$all" | grep -c .)"
  n_tasks="$(printf '%s\n' "$all" | awk 'NF>=2{print $1}' | sort -u | grep -c .)"
  printf '        重投后 ⇒ 回执 %s 条(原 %s)，涉及 task_id %s 个\n' "${n_recv2:-0}" "${n_recv:-0}" "${n_tasks:-0}"
  if [ "${n_recv2:-0}" -le "${n_recv:-0}" ]; then
    bad "重投后没有任何新回执 ⇒ 这条断言其实没测到东西（不许当成通过）"; return 1
  fi
  if [ "${n_tasks:-0}" -eq 1 ]; then
    ok "重投同一条事件仍只有**一个** task_id(R-1 幂等成立：不重复执行)"
  else
    bad "出现 ${n_tasks} 个 task_id ⇒ 幂等在真链路上失效"; printf '%s\n' "$all" | sed 's/^/        /'; return 1
  fi
  return 0
}

# 从 /messages 取指定 thread root 下的回执：每行 "<task_id> <status>"；取数失败给哨兵
probe_receipts() { # probe_receipts "<root event ids...>"
  local f="$TMP/pr.json"
  api GET "/_matrix/client/v3/rooms/$(enc "$ROOM")/messages?dir=b&limit=100" > "$f" || { echo __FETCH_FAILED__; return 1; }
  python3 -c '
import json, sys
path = sys.argv[1]; roots = set(sys.argv[2].split())
try:
    d = json.load(open(path, encoding="utf-8"))
except Exception:
    print("__FETCH_FAILED__"); sys.exit(0)
if not isinstance(d, dict) or "chunk" not in d:
    print("__FETCH_FAILED__"); sys.exit(0)
rows = []
for ev in (d.get("chunk") or []):
    c = ev.get("content") or {}
    rel = c.get("m.relates_to") or {}
    root = rel.get("event_id") if rel.get("rel_type") == "m.thread" else None
    # 真回执把 task_id/status 放在 content **顶层**（事件类型才是 com.apescale.…result.v1）；
    # 离线替身里也出现过嵌套形态 ⇒ 两种都认，别只按一种写死（我就因此漏抓过一次）。
    r = c.get("com.apescale.tutti.result.v1") or c
    tid, st = r.get("task_id"), r.get("status")
    if tid and st and root in roots:
        rows.append((int(ev.get("origin_server_ts") or 0), tid, st))
for _, tid, st in sorted(rows):
    print(tid, st)
' "$f" "$1"
}

selftest() {
  local out port pid rc
  out="$(mktemp)"
  echo "── 0. 起 mock matrix 替身"
  MOCK_TOKEN=mock-token python3 "$(dirname "$0")/../tests/mock_matrix.py" 0 >"$out" 2>&1 &
  pid=$!
  trap 'kill "${pid:-}" 2>/dev/null; rm -rf "${TMP:-}" "${out:-}"' EXIT
  for _ in $(seq 1 50); do port="$(awk '/LISTENING/{print $2; exit}' "$out" 2>/dev/null)"; [ -n "${port:-}" ] && break; sleep 0.1; done
  [ -z "${port:-}" ] && { echo "mock 未启动"; cat "$out"; exit 1; }
  echo "   ✅ 监听 127.0.0.1:$port"

  export MATRIX_BASE_URL="http://127.0.0.1:$port" MATRIX_ACCESS_TOKEN=mock-token
  export BRIDGE_ROOM='!p4:matrix.test' NODE_NAME=selftest P4_WAIT=6 P4_NEG_WAIT=1
  export TARGET_A='@tutti-hermes:matrix.test' TARGET_B='@tutti-pi:matrix.test'
  export BRIDGE_A='@bridge3:matrix.test' BRIDGE_B='@bridge2:matrix.test'

  echo "── 1. 正例:应全绿(exit 0)"
  run_live >/dev/null 2>&1; rc=$?
  [ "$rc" -eq 0 ] && echo "   ✅ 正例通过" || { echo "   ❌ 正例失败 exit=$rc"; run_live | sed 's/^/      /'; exit 1; }

  echo "── 2. 对照:缺 running(R-3 违例) 必须判红(exit 1)"
  kill $pid 2>/dev/null; wait $pid 2>/dev/null
  MOCK_TOKEN=mock-token MOCK_SKIP_RUNNING=1 python3 "$(dirname "$0")/../tests/mock_matrix.py" 0 >"$out" 2>&1 &
  pid=$!
  for _ in $(seq 1 50); do port="$(awk '/LISTENING/{print $2; exit}' "$out" 2>/dev/null)"; [ -n "${port:-}" ] && break; sleep 0.1; done
  export MATRIX_BASE_URL="http://127.0.0.1:$port"
  run_live >/dev/null 2>&1; rc=$?
  [ "$rc" -eq 1 ] && echo "   ✅ 缺 running 时门正确报红(exit 1)" || { echo "   ❌ 缺 running 未报红 exit=$rc"; exit 1; }

  echo "── 3. 对照:房间缺第二个 target 必须判前置不足(exit 2)"
  kill $pid 2>/dev/null; wait $pid 2>/dev/null
  MOCK_TOKEN=mock-token MOCK_NO_TARGET_B=1 python3 "$(dirname "$0")/../tests/mock_matrix.py" 0 >"$out" 2>&1 &
  pid=$!
  for _ in $(seq 1 50); do port="$(awk '/LISTENING/{print $2; exit}' "$out" 2>/dev/null)"; [ -n "${port:-}" ] && break; sleep 0.1; done
  export MATRIX_BASE_URL="http://127.0.0.1:$port"
  run_live >/dev/null 2>&1; rc=$?
  [ "$rc" -eq 2 ] && echo "   ✅ 缺 target 时拒绝判读(exit 2)" || { echo "   ❌ 缺 target 未判前置不足 exit=$rc"; exit 1; }

  echo "── 4. 对照:另一节点的桥抢答(串台, R-7 违反) 必须判红(exit 1)"
  kill $pid 2>/dev/null; wait $pid 2>/dev/null
  MOCK_TOKEN=mock-token MOCK_LEAK=1 python3 "$(dirname "$0")/../tests/mock_matrix.py" 0 >"$out" 2>&1 &
  pid=$!
  for _ in $(seq 1 50); do port="$(awk '/LISTENING/{print $2; exit}' "$out" 2>/dev/null)"; [ -n "${port:-}" ] && break; sleep 0.1; done
  export MATRIX_BASE_URL="http://127.0.0.1:$port" MATRIX_ACCESS_TOKEN=mock-token
  run_live >/dev/null 2>&1; rc=$?
  [ "$rc" -eq 1 ] && echo "   ✅ 串台时门正确报红(exit 1)" || { echo "   ❌ 串台未报红 exit=$rc"; exit 1; }

  echo "── 5. 对照:错 token 必须拒绝判读(exit 2)"
  export MATRIX_ACCESS_TOKEN=wrong-token
  run_live >/dev/null 2>&1; rc=$?
  [ "$rc" -eq 2 ] && echo "   ✅ 令牌失效时拒绝判读(exit 2)" || { echo "   ❌ 错 token 未拒绝 exit=$rc"; exit 1; }
  kill $pid 2>/dev/null

  echo "── 6. --pair:两台机各跑一次的原始输出应被**机械**判定为通过"
  kill $pid 2>/dev/null; wait $pid 2>/dev/null
  MOCK_TOKEN=mock-token python3 "$(dirname "$0")/../tests/mock_matrix.py" 0 >"$out" 2>&1 &
  pid=$!
  for _ in $(seq 1 50); do port="$(awk '/LISTENING/{print $2; exit}' "$out" 2>/dev/null)"; [ -n "${port:-}" ] && break; sleep 0.1; done
  export MATRIX_BASE_URL="http://127.0.0.1:$port" MATRIX_ACCESS_TOKEN=mock-token
  run_live > "$TMP/pa.log" 2>&1
  run_live > "$TMP/pb.log" 2>&1
  sed -i.bak 's/node=selftest/node=mac13/' "$TMP/pa.log"                       # 只改"我在哪台机"
  sed -i.bak 's/node=selftest/node=mac15/; s/user=@bridge3:matrix.test/user=@bridge2:matrix.test/' "$TMP/pb.log"
  if pair_check "$TMP/pa.log" "$TMP/pb.log"; then
    echo "   ✅ 两节点/两桥/同房间/归属唯一且跨轮一致 ⇒ 机械判定通过"
  else
    echo "   ❌ --pair 正例失败"; exit 1
  fi

  echo "── 7. 对照:两侧 node 相同(同一台机跑两次) 必须判红"
  cp "$TMP/pa.log" "$TMP/pc.log"
  if pair_check "$TMP/pa.log" "$TMP/pc.log" >/dev/null 2>&1; then
    echo "   ❌ 同一节点跑两次竟被判通过"; exit 1
  else
    echo "   ✅ 同一节点被判红(确实不是两台物理机)"
  fi

  echo "── 8. 对照:跨轮归属不一致(同一 target 换桥应答) 必须判红 —— R-7"
  sed 's/target=@tutti-hermes:matrix.test bridge=@bridge3:matrix.test/target=@tutti-hermes:matrix.test bridge=@bridge9:matrix.test/' \
      "$TMP/pb.log" > "$TMP/pd.log"
  if pair_check "$TMP/pa.log" "$TMP/pd.log" >/dev/null 2>&1; then
    echo "   ❌ 跨轮归属不一致竟被判通过"; exit 1
  else
    echo "   ✅ 归属不稳定被判红(R-7)"
  fi

  echo "── 9. 对照:两侧 target 零交叉(各只测一个) 必须判红 —— 缺陷K"
  grep -v "tutti-pi" "$TMP/pa.log" > "$TMP/pe.log"       # A 侧只留 hermes
  grep -v "tutti-hermes" "$TMP/pb.log" > "$TMP/pf.log"   # B 侧只留 pi
  if pair_check "$TMP/pe.log" "$TMP/pf.log" >/dev/null 2>&1; then
    echo "   ❌ 零交叉竟被判通过(阶段门放水)"; exit 1
  else
    echo "   ✅ 零交叉被判红(缺陷K 已修)"
  fi

  echo "── 10. 对照:/messages 取数 403 ⇒ 阴性对照必须报取数失败 —— 缺陷L"
  kill $pid 2>/dev/null; wait $pid 2>/dev/null
  MOCK_TOKEN=mock-token MOCK_MESSAGES_403=1 python3 "$(dirname "$0")/../tests/mock_matrix.py" 0 >"$out" 2>&1 &
  pid=$!
  for _ in $(seq 1 50); do port="$(awk '/LISTENING/{print $2; exit}' "$out" 2>/dev/null)"; [ -n "${port:-}" ] && break; sleep 0.1; done
  export MATRIX_BASE_URL="http://127.0.0.1:$port" MATRIX_ACCESS_TOKEN=mock-token
  run_live >/dev/null 2>&1; rc=$?
  [ "$rc" -eq 2 ] && echo "   ✅ 取数失败时拒绝判读(exit 2), 不再 fail-open" \
    || { echo "   ❌ 取数失败未拒绝判读 exit=$rc"; exit 1; }

  echo
  echo "═══ 15-criterion4-live selftest: 全部通过（正例 2 + 对照 8）═══"
}

case "${1:-}" in
  --selftest) selftest;;
  --pair) pair_check "${2:-}" "${3:-}";;
  ""|--live) run_live;;
  --help|-h) sed -n '2,28p' "$0";;
  *) echo "未知参数: $1（用 --selftest 或 --help）" >&2; exit 2;;
esac