#!/usr/bin/env python3
"""Shared POG422 fixture instrumentation — handshake, serial, events, packet parse."""

from __future__ import annotations

import argparse
import csv
import fcntl
import glob
import itertools
import json
import os
import select
import serial
import shutil
import socket
import struct
import subprocess
import sys
import tempfile
import threading
import time
import zlib
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterator, Optional, Sequence

SOH = 0x05
EOP = 0x0D

# Machine-bus SOH frame layout (every opcode):
#   [0]=SOH  [1]=opcode  [2]=BANK_NUMBER  [3..4]=TID LE (most msgs)  …
# Bank Number is the cabinet bank wire id (not TID). Filter/stamp everywhere.
POG422_BANK_NUMBER_OFFSET = 2

# EtherGold product BOM — MajorFeature.MinorFeature.MajorFix.MinorFix (see EG_CHANGELOG.md).
# EG_VERSION = the whole comm aggregate (HTTP, listen, pogcc-com, Python, OTA).
# Any of those component changes bumps this stamp. It is what OTA / /health show.
# UARTD_ABI_VERSION = last uartd/inject ELF rebuild. Bump only when those binaries
# change (see update.sh + probe_component_versions). Not a second product.
# XR driver may still lag within [EG_COMPONENT_MIN_VERSION, MAX] until ABI requires a bump.
EG_VERSION = "4.1.0.26"
EG_VERSION_SLUG = "dev-skip-auto-promote"
UARTD_ABI_VERSION = "4.1.0.11"

# XR driver floor..tip. Raise MIN when this tree *requires* a newer driver ABI.
EG_COMPONENT_MIN_VERSION = "2.2.2.12"  # EXTERNAL_DE return-after-bulk (no kernel air usleep).
EG_COMPONENT_MAX_VERSION = EG_VERSION

# parser_build keeps a human slug; match logic uses EG_VERSION.
FIXTURE_PARSER_BUILD = f"eg-{EG_VERSION}-{EG_VERSION_SLUG}"
FIXTURE_PARSER_BUILD_PROD = f"eg-prod-{EG_VERSION}-{EG_VERSION_SLUG}"

# Loaded XR module compatibility (logged at session_start).
XR_DRIVER_MODULE = "xr_usb_serial_common"
# Driver: range check. UARTD/inject: exact UARTD_ABI_VERSION (not EG_VERSION).
EXPECTED_XR_DRIVER_VERSION = EG_COMPONENT_MAX_VERSION
EXPECTED_UARTD_VERSION = UARTD_ABI_VERSION
EXPECTED_INJECT_VERSION = UARTD_ABI_VERSION
# Legacy non A.B.C.D module IDs still seen on some benches (not range-comparable).
ACCEPTABLE_XR_DRIVER_LEGACY = frozenset(
    {
        "1W-pog422-tx-fifo-errno-ch",
        "1V-pog422-tx-fifo-dma-buf",
        "1U-pog422-tx-fifo-uart-block",
        "1T-pog422-tx-fifo-get-reg-fix",
        "1S-pog422-inject-tx-fifo-diag",
        "1R-pog422-preciseflags-ioctl-fix",
    }
)
# Back-compat alias for older tools that imported the frozenset name.
ACCEPTABLE_XR_DRIVER_VERSIONS = ACCEPTABLE_XR_DRIVER_LEGACY

# After eth→UART x9 inject, machine x1 often lands before the next GPIO poll.
# EG 3.0.1.54: restore pre-.53 local grace (25 ms); .53 20 ms hunt = continuous MRT.
POST_X9_INJECT_FIFO_GRACE_MS = 15.0
POST_X9_INJECT_LOCAL_TID_FIFO_GRACE_MS = 45.0
# After local x9: warn if machine x1 / closing 0x0D miss the MRT window (49→41→0D).
MRT_RISK_MISSING_X1_MS = 3000.0
# EG 3.0.1.54: restore .52 late-0D wait (150 ms). .53 cut to 20 → glass MRT.
MRT_RISK_LATE_0D_MS = 150.0
# EG 3.0.1.48/49: same-seq local wager may reopen *ok* close-done only after this
# age (soak 1513: .47 reopened on CGS re-wager ~1 ms after good uartd close).
# .48 used 2500 ms and restored continuous seq_already_closed latch (glass re-Play
# before settle). .49 keeps a short OK settle; recovery/miss closes reopen immediately.
MRT_HAND_REOPEN_MIN_AGE_MS = 400.0
# After armed local x9: if Master does not re-send same-seq x9 within this window,
# *historical* path inferred x1 + synth 0D. EG 3.0.1.55: infer disabled — empty
# harvest leaves hand open for real Master re-49 / x1 (constants retained for logs).
MRT_X1_INFER_NO_X9_RETRY_MS = 25.0
# If Master *did* re-send same-seq x9 (x9_retry_seen) but local x1 still missing —
# same short timer used only for diagnostics while hand stays open (.55).
MRT_X1_INFER_AFTER_X9_RETRY_MS = 25.0
# EG 4.0.0.71 diag: after empty peel (peel_src=0), keep tagging copper RX for this
# long so a late Slave 81 can be proven in events without changing product path.
POST_X9_ORPHAN_RX_MS = 300.0
# Hub: same-TID Master copper 89 within this gap after prior 89 → retry crumb.
MRT_HUB_X9_RETRY_MIN_GAP_MS = 50.0
MRT_HUB_X9_RETRY_MAX_GAP_MS = 400.0
# EG 3.0.1.45 OPEN GUARD was REVERTED in 3.0.1.46 (soak master-no-slave-cgs):
# serializing foreign 0x40 while any trunk open deadlocked multi-seat Master —
# deferred queue filled to cap, Master copper got no slave wagers, slaves CGS.
# Do not re-introduce _any_trunk serialize-on-open without a flush/timeout proof.
# MASTER_DEAL_OPEN_GUARD_MS kept as documentation only (unused).
MASTER_DEAL_OPEN_GUARD_MS = 140.0  # historical; not wired after 3.0.1.46

# EG 3.0.1.44: brief wait for late AUTO_CLOSE_DONE only (not a glass MRT fix).
# EG 3.0.1.50 stretched this to 250–350 ms — does not prevent glass MRT and can
# delay real closes; .51 keeps short defer and relies on local game-txn exclusive.
STATUS_0D_PYTHON_FALLBACK_DEFER_MS = 20.0
# Unused after .51 exclusive path; kept so field configs/tests resolving the name
# do not break. Prefer local_game_txn exclusive + supersede on CGS, not long wait.
STATUS_0D_PYTHON_FALLBACK_DEFER_CGS_MS = 20.0
# Hard cap for an exclusive local game transaction (x0→0D). Stale-abort only.
LOCAL_GAME_TXN_MAX_MS = 2000.0
# After local-TID x9 inject: hold *non-local* eth→UART briefly so the T340+ can
# ACK without us re-grabbing the bus (witness soak: x1 ~0.8s after x9 under load).
# Soft hold only — at MRT_RISK_MISSING_X1_MS the bridge abandons await_x1 + trunk.
MRT_LOCAL_X1_INJECT_HOLD_MS = 2000.0
# Cap foreign frames stashed behind transaction gate (overnight soak hit 1000+).
# EG 3.0.2.7: 64 → 128; spoke peer-gameplay no longer queues for eth→UART, but
# soft/status traffic under exclusive can still pulse above 64 in multi-seat EDGE.
TRANSACTION_DEFER_MAX_QUEUE = 128
# Open trunk without close — abandon so game_transaction_gate / 0A cannot stick.
TRUNK_ROUND_STALE_MS = 8000.0
# inject_deferred / transaction_defer are on the hot eth_rx path — never log every tick.
INJECT_DEFER_LOG_MIN_INTERVAL_MS = 1000.0
# EG 2.2.2.69: gameplay eth→UART / synth 0D must not pin _tx_lock for the legacy
# 5 s IPC hang (pos5 18:33 — inject_total_us≈5e6). USB chunk timeout is ~500 ms;
# allow one margin. Non-gameplay keeps the long hang ceiling.
# EG 2.2.2.70: budget is echo-armed — once echo/post_DE/forensics prove the TX,
# uart_client extends once (INJECT_DONE_EXTEND_S) for a late DONE instead of
# failing a delivered inject (pos5 00:09 MRT).
INJECT_WAIT_S_GAMEPLAY = 0.55
# EG 3.0.1.15: was 5.0 — that budget matched the USB control 5s timeout and left
# DE/HS asserted for a full five seconds on meter 0x0A (pos5 22:44). IPC wait is
# not a DE budget; fail fast and let abort / driver hard-cap clear the line.
INJECT_WAIT_S_DEFAULT = 0.75
# EG 2.2.2.71: eth fan-out wall-time study (hub scale); log when ports >= this.
ETH_FANOUT_LOG_MIN_PORTS = int(os.environ.get("ETH_FANOUT_LOG_MIN_PORTS", "2"))
# TID-aware unicast on EDGE hub (default on). Flood when TID→port unknown.
EDGE_TID_UNICAST = os.environ.get("EDGE_TID_UNICAST", "1").strip().lower() not in (
    "0",
    "false",
    "no",
    "off",
)
# Slow-inject log threshold (µs) — matches prior mrt_x9_inject_slow gate.
INJECT_SLOW_LOG_US = 100_000.0

# Prod errors.jsonl rotation (default 10 KiB; FIXTURE_PROD_ERRORS_MAX_BYTES=0 disables).
PROD_ERRORS_MAX_BYTES = int(os.environ.get("FIXTURE_PROD_ERRORS_MAX_BYTES", "10240"))
PROD_LOG_MAX_AGE_S = float(os.environ.get("FIXTURE_PROD_LOG_MAX_AGE_S", "86400"))

# Debug capture rotation — events.jsonl + packets.raw rotate together (~4–5 h at soak rates).
DEBUG_LOG_DEFAULT_MAX_BYTES = 25 * 1024 * 1024
DEBUG_LOG_DEFAULT_PRUNE_AGE_S = 7 * 86400


def _env_bool(name: str, default: bool) -> bool:
    raw = os.environ.get(name, "").strip().lower()
    if not raw:
        return default
    return raw in ("1", "true", "yes", "on")


def _env_int(name: str, default: int) -> int:
    raw = os.environ.get(name, "").strip()
    if not raw:
        return default
    try:
        return int(raw, 0)
    except ValueError:
        return default


def _env_float(name: str, default: float) -> float:
    raw = os.environ.get(name, "").strip()
    if not raw:
        return default
    try:
        return float(raw)
    except ValueError:
        return default


def debug_log_rotate_enabled(*, prod_mode: bool) -> bool:
    if prod_mode:
        return False
    return _env_bool("FIXTURE_DEBUG_LOG_ROTATE", True)


def debug_log_max_bytes() -> int:
    return _env_int("FIXTURE_DEBUG_LOG_MAX_BYTES", DEBUG_LOG_DEFAULT_MAX_BYTES)


def debug_log_prune_age_s() -> float:
    return _env_float("FIXTURE_DEBUG_LOG_MAX_AGE_S", DEBUG_LOG_DEFAULT_PRUNE_AGE_S)


def debug_log_utc_daily_enabled() -> bool:
    return _env_bool("FIXTURE_DEBUG_LOG_ROTATE_UTC_DAILY", True)


def _debug_log_utc_stamp() -> str:
    return time.strftime("%Y%m%dT%H%M%SZ", time.gmtime())


def _file_size_or_zero(path: Path) -> int:
    if not path.is_file():
        return 0
    try:
        return int(path.stat().st_size)
    except OSError:
        return 0


def _capture_file_sizes(output_dir: Path) -> tuple[int, int, int]:
    events_bytes = _file_size_or_zero(output_dir / EVENTS_JSONL)
    packets_bytes = _file_size_or_zero(output_dir / PACKETS_RAW)
    echo_bytes = _file_size_or_zero(output_dir / INJECT_ECHO_JSONL)
    return events_bytes, packets_bytes, echo_bytes


def _debug_log_rotate_reason(
    output_dir: Path,
    *,
    max_bytes: int,
    utc_daily: bool,
    utc_day: str,
) -> tuple[str, int, int] | None:
    events_bytes, packets_bytes, echo_bytes = _capture_file_sizes(output_dir)
    if events_bytes <= 0 and packets_bytes <= 0 and echo_bytes <= 0:
        return None
    if max_bytes > 0 and (
        events_bytes >= max_bytes
        or packets_bytes >= max_bytes
        or echo_bytes >= max_bytes
    ):
        return "size", events_bytes, packets_bytes
    if utc_daily:
        now_day = datetime.now(timezone.utc).strftime("%Y-%m-%d")
        if utc_day != now_day:
            return "utc_day", events_bytes, packets_bytes
    return None


def prune_debug_rotated_logs(output_dir: Path, *, max_age_s: float) -> int:
    """Delete events.jsonl.* / packets.raw.* older than max_age_s."""
    if max_age_s <= 0 or not output_dir.is_dir():
        return 0
    cutoff = time.time() - max_age_s
    removed = 0
    for base in (EVENTS_JSONL, PACKETS_RAW, INJECT_ECHO_JSONL, HW_DEBUG_JSONL):
        prefix = f"{base}."
        for entry in output_dir.iterdir():
            if not entry.is_file() or not entry.name.startswith(prefix):
                continue
            try:
                if entry.stat().st_mtime < cutoff:
                    entry.unlink()
                    removed += 1
            except OSError:
                pass
    return removed


def rotate_debug_capture_logs(
    output_dir: Path,
    *,
    max_bytes: int,
    utc_daily: bool,
    utc_day: str,
) -> dict[str, Any] | None:
    """Rotate events.jsonl, packets.raw, and inject_echo.jsonl with one UTC stamp."""
    reason_info = _debug_log_rotate_reason(
        output_dir,
        max_bytes=max_bytes,
        utc_daily=utc_daily,
        utc_day=utc_day,
    )
    if reason_info is None:
        return None
    reason, events_bytes, packets_bytes = reason_info
    stamp = _debug_log_utc_stamp()
    archived: dict[str, str] = {}
    for base in (EVENTS_JSONL, PACKETS_RAW, INJECT_ECHO_JSONL, HW_DEBUG_JSONL):
        src = output_dir / base
        if not src.is_file():
            continue
        dest = output_dir / f"{base}.{stamp}"
        n = 1
        while dest.exists():
            dest = output_dir / f"{base}.{stamp}.{n}"
            n += 1
        src.rename(dest)
        archived[base] = dest.name
    if not archived:
        return None
    prune_debug_rotated_logs(output_dir, max_age_s=debug_log_prune_age_s())
    return {
        "reason": reason,
        "events_bytes": events_bytes,
        "packets_bytes": packets_bytes,
        "archive_events": archived.get(EVENTS_JSONL),
        "archive_packets": archived.get(PACKETS_RAW),
        "archive_inject_echo": archived.get(INJECT_ECHO_JSONL),
    }

# 0x0A progressive broadcasts carry tid=0000 (bank entity tail) — not a terminal slot ID.
BANK_PROGRESSIVE_BROADCAST_TID = "0000"

# Wire byte count (SOH through EOP inclusive) per command code at byte 1.
# Gameplay: 2026-06-11_2045 twelve-games + later censuses.
# STOM Transfer (any Master, locked 2026-07-13): Full 01@32→09@1170; Partial 04@32→0E@598.
#   See PROTOCOL_TRANSACTIONS.md §1b. Keep UARTD pog422_wire_len*_ in sync.
# Payload may contain 0x0D — never hunt EOP when a known length matches.
COMMAND_WIRE_LEN: dict[int, int] = {
    0x01: 17,  # draw poker skill grant (Master → slave; SOH path; bench 2026-07-02)
    0x02: 32,  # slave auto-enroll / update request (POG_UPDT — not Partial menu)
    0x03: 32,  # SCR
    0x04: 32,  # Partial Transfer request (any Master)
    0x0E: 598,  # Partial Transfer response (any Master)
    0x09: 1170,  # Full Transfer / POG_CONF response (any Master; alt 1094 legacy)
    0x0A: 82,  # progressive broadcast
    0x0B: 12,  # master enroll ack
    0x0D: 14,  # status/accounting
    0x20: 37,  # Touch 6 wager (primary, 1-card)
    0x21: 12,  # Touch 6 ack
    0x29: 30,  # Touch 6 result
    0x30: 80,  # SuperGold Bingo wager (T340+ fixed four-slot frame; bench 2026-07-04)
    0x31: 12,  # Bingo ack
    0x39: 74,  # Bingo result
    0x40: 42,  # keno wager
    0x41: 12,  # keno ack
    0x49: 44,  # keno result
    0x50: 32,  # draw poker wager
    0x51: 12,  # draw poker ack
    0x52: 32,  # draw poker skill jpot claim (slave → master; bench 2026-07-02)
    0x59: 34,  # draw poker result (primary)
    0x5A: 24,  # draw poker skill jpot grant (bench 2026-07-02 copper; was 20 TBD)
    0x60: 48,  # SuperPick wager
    0x61: 12,  # SuperPick ack
    0x69: 30,  # SuperPick result (primary)
    0x70: 32,  # blackjack wager
    0x71: 12,  # blackjack ack
    0x72: 32,  # blackjack / SpinJack skill jpot claim (slave → master; bench 2026-07-03)
    0x79: 42,  # blackjack result (longest common)
    0x7A: 24,  # blackjack / SpinJack skill jpot grant (parallel 0x5A; inferred +0x20 family)
    0x80: 32,  # 8-line slot wager
    0x81: 12,  # slot ack
    0x89: 36,  # slot result
}

# Alternate wire lengths (same opcode). Tried longest-first after primary.
COMMAND_WIRE_LEN_ALTS: dict[int, tuple[int, ...]] = {
    # Slave STOM config request (POG_CONF) shares opcode 0x01 with Master MTOS poker grant @ 17 B.
    0x01: (32,),
    # Legacy comlengs[9] / synthetic self-test length (kept if some Masters emit it).
    0x09: (1094,),
    # Jun 19 census mid-window EOP (looks like 397); live Partial reply is 598.
    0x0E: (397,),
    # 58 B = 1835 AF1D census (two sequential 12 B cards). T340+ uses 80 B for all PLAY counts.
    0x30: (58,),
    0x20: (48,),  # 2+ cards (1859 card-matrix); T340+ never emits 41 B
    0x59: (23,),  # short poker result
    0x5A: (20,),  # legacy doc length; bench copper is 24 B
    0x7A: (20,),  # legacy doc length; parallel to 0x5A family
    0x69: (28,),  # short SuperPick result
    0x79: (38, 34, 31),  # blackjack mid-hand variants
}


def wire_lengths_for_cmd(cmd: int) -> tuple[int, ...]:
    """Known wire lengths for cmd, longest first (for split_packets)."""
    primary = COMMAND_WIRE_LEN.get(cmd)
    alts = COMMAND_WIRE_LEN_ALTS.get(cmd, ())
    seen: set[int] = set()
    out: list[int] = []
    for n in ([primary] if primary is not None else []) + list(alts):
        if n is not None and n not in seen:
            seen.add(n)
            out.append(n)
    out.sort(reverse=True)
    return tuple(out)


def wire_len_valid(raw: bytes) -> bool:
    """True when frame length matches COMMAND_WIRE_LEN* for its opcode."""
    if len(raw) < 2 or raw[0] != SOH or raw[-1] != EOP:
        return False
    return len(raw) in wire_lengths_for_cmd(raw[1])


def short_frame_bytes_needed(buf: bytes) -> tuple[int, int] | None:
    """SOH + known opcode, not yet a COMMAND_WIRE_LEN packet → (opcode, still_need).

    GPIO fall means copper DE is done, not that USB has delivered the frame.
    Hold until wire length (not interior EOP), next GPIO rise, or SHORT_FRAME_HOLD_NS.
    """
    probe = strip_before_mark(buf)
    if not probe or len(probe) < 2 or probe[0] != SOH:
        return None
    opcode = probe[1]
    known = wire_lengths_for_cmd(opcode)
    if not known:
        return None
    if split_packets(probe):
        return None
    max_n = max(known)
    if len(probe) >= max_n:
        return None
    return opcode, max_n - len(probe)


def extract_valid_local_x1_ack_frames(
    buf: bytes,
    local_tid_hex: str | None = None,
) -> list[bytes]:
    """Wire-len-valid local gameplay ACK packets from a bus RX buffer.

    Rejects first-EOP false frames (EOP==0x0D yields 6 B SOH..junk that is not
    a real 12 B ack). Mirrors uartd extract_local_x1_frame (EG 3.0.1.21).
    """
    if not buf:
        return []
    want_tid = (local_tid_hex or "").strip().upper().replace(" ", "")
    out: list[bytes] = []
    normalized = normalize_window(buf)
    if normalized:
        for pkt in split_packets(normalized):
            if not wire_len_valid(pkt):
                continue
            if not is_gameplay_ack(pkt[1]):
                continue
            if want_tid:
                tid = f"{pkt[3]:02X}{pkt[4]:02X}"
                if tid != want_tid:
                    continue
            out.append(pkt)
        return out
    # Fallback: fixed 12 B ack scan (no normalize) — same as uartd.
    ack_len = 12
    i = 0
    while i + ack_len <= len(buf):
        if buf[i] != SOH:
            i += 1
            continue
        if not is_gameplay_ack(buf[i + 1]):
            i += 1
            continue
        if buf[i + ack_len - 1] != EOP:
            i += 1
            continue
        pkt = bytes(buf[i : i + ack_len])
        if not wire_len_valid(pkt):
            i += 1
            continue
        if want_tid:
            tid = f"{pkt[3]:02X}{pkt[4]:02X}"
            if tid != want_tid:
                i += 1
                continue
        out.append(pkt)
        i += ack_len
    return out


def parser_self_test() -> list[str]:
    """Return error strings for install.sh (empty = pass)."""
    errors: list[str] = []

    def _pkt(csv: str) -> bytes:
        return bytes(int(b, 16) for b in csv.split(","))

    scr32 = bytes(
        [
            0x05,
            0x03,
            0x01,
            0x37,
            0xBC,
            0x0C,
            0x97,
            0x01,
            0x01,
            0x00,
            0x00,
            0x00,
            0x00,
            0x00,
            0x00,
            0x00,
            0x00,
            0xB2,
            0x0D,
            0x04,
            0x00,
            0x00,
            0x00,
            0x00,
            0x00,
            0x00,
            0x00,
            0x00,
            0x00,
            0xB4,
            0x94,
            0x0D,
        ]
    )
    if len(scr32) != 32:
        errors.append(f"scr32 test vector len {len(scr32)}")
    if parse_packet_meta(scr32).tid_hex != "37BC":
        errors.append("terminal tid from SCR/config (0x03)")
    parts = split_packets(scr32)
    if len(parts) != 1 or len(parts[0]) != 32:
        errors.append(f"split_packets 0x03 wire len: {[len(p) for p in parts]!r}")
    if not is_stom_comm_priority(SCR_REPLY):
        errors.append("is_stom_comm_priority must include SCR_REPLY (0x0B)")
    if not is_stom_comm_priority(0x0A):
        errors.append("is_stom_comm_priority must include POG_UPDT reply 0x0A")
    if not is_stom_comm_request(0x02) or not is_stom_comm_priority(0x09):
        errors.append("STOM must include T340+ 0x02 request and Master 0x09")
    if eth_only_ports_unmapped(0x02, topo="hub") is not None:
        errors.append("hub unmapped 0x02 must flood (not [])")
    if eth_only_ports_unmapped(0x40, topo="hub") != []:
        errors.append("hub unmapped cash x0 still nowhere")
    if eth_only_ports_unmapped(0x40, topo="spoke") is not None:
        errors.append("spoke unmapped cash must flood uplink")
    # EG 4.1.0.1: SCR is not a bank broadcast — hub unmapped must not multi-spoke flood.
    if eth_only_ports_unmapped(SCR_MSG, topo="hub") != []:
        errors.append("hub unmapped SCR 0x03 must be nowhere ([])")
    if eth_only_ports_unmapped(SCR_REPLY, topo="hub") != []:
        errors.append("hub unmapped SCR reply 0x0B must be nowhere ([])")
    if eth_only_ports_unmapped(SCR_MSG, topo="spoke") is not None:
        errors.append("spoke unmapped SCR must flood uplink")
    if eth_only_ports_unmapped(SCR_REPLY, topo="spoke") is not None:
        errors.append("spoke unmapped SCR reply must flood uplink")
    scr_reply12 = bytes.fromhex("050B0138D70C9704000DD70D")
    glued_scr = scr32 + scr_reply12
    ok, kind = inject_echo_bytes_match(scr32, glued_scr)
    if not ok or kind != "prefix":
        errors.append(f"inject_echo_bytes_match glued SCR+0B: {ok=!r} {kind=!r}")
    ok_exact, kind_exact = inject_echo_bytes_match(scr32, scr32)
    if not ok_exact or kind_exact != "exact":
        errors.append(f"inject_echo_bytes_match exact SCR: {ok_exact=!r} {kind_exact=!r}")
    peer = peer_uart_bytes_after_inject(scr32, glued_scr, b"")
    if peer != scr_reply12:
        errors.append(f"peer_uart_bytes_after_inject glued SCR+0B: {peer!r}")
    if peer_uart_bytes_after_inject(scr32, scr32, b"") != b"":
        errors.append("peer_uart_bytes_after_inject exact SCR must be empty")
    if peer_uart_bytes_after_inject(scr32, b"", scr_reply12) != scr_reply12:
        errors.append("peer_uart_bytes_after_inject post_de-only peer")
    # Lone post-DE SOH of a forthcoming x1 must remain peelable as peer bytes
    # (bridge holds it — must not vanish in peer_uart).
    if peer_uart_bytes_after_inject(scr32, scr32, bytes([SOH])) != bytes([SOH]):
        errors.append("peer_uart_bytes_after_inject lone post_de SOH")
    # Mismatch with embedded self-echo then glued peer (pos4 A1 / Class D).
    # Bidirectional peel keeps junk prefix + trailing peer (EG 2.2.2.69).
    junk_prefix = bytes([0x89, 0x80, 0x00, 0x00])
    peer81 = bytes([SOH, 0x81] + [0x00] * 9 + [EOP])  # 0x81 wire_len=12
    mismatch_bus = junk_prefix + scr32 + peer81
    peel_mm = peer_uart_bytes_after_inject(scr32, mismatch_bus, b"")
    if peel_mm != junk_prefix + peer81:
        errors.append(
            f"peer_uart_bytes_after_inject mismatch embedded sent: {peel_mm!r}"
        )
    ok_emb, kind_emb = inject_echo_bytes_match(scr32, mismatch_bus)
    if not ok_emb or kind_emb != "embedded":
        errors.append(
            f"inject_echo_bytes_match embedded: {ok_emb=!r} {kind_emb=!r}"
        )
    # Leading peer x1 then self x9 (pos5 18:33 EDT class).
    peer61 = bytes([SOH, 0x61] + [0x00] * 9 + [EOP])  # 12 B ack
    x9_30 = bytes([SOH, 0x69] + [0x00] * 27 + [EOP])
    leading61 = peer61 + x9_30
    peel_lead = peer_uart_bytes_after_inject(x9_30, leading61, b"")
    if peel_lead != peer61:
        errors.append(
            f"peer_uart_bytes_after_inject leading 61|x9: {peel_lead!r}"
        )
    ok_lead, kind_lead = inject_echo_bytes_match(x9_30, leading61)
    if not ok_lead or kind_lead != "embedded":
        errors.append(
            f"inject_echo_bytes_match leading 61|x9: {ok_lead=!r} {kind_lead=!r}"
        )
    trail61 = x9_30 + peer61
    if peer_uart_bytes_after_inject(x9_30, trail61, b"") != peer61:
        errors.append("peer_uart_bytes_after_inject trailing x9|61")

    # Truncated SCR tail — interior/final 0x0D is not a valid 32 B frame; suppress crop.
    stub6 = bytes.fromhex("05030137BC0D")
    parts = split_packets(stub6)
    if parts:
        errors.append(f"split_packets truncated SCR must not crop: {parts!r}")

    # Interior 0x0D must not produce a short false 0x79 before the real 42 B frame.
    bj_79_42 = _pkt(
        "05,79,01,A4,5B,0F,96,00,00,00,01,86,A0,00,00,00,00,2A,00,6B,8C,3B,11,0B,"
        "14,09,25,1D,04,2A,38,33,17,3C,0A,3D,06,07,2D,B5,FC,0D"
    )
    false79 = bytes([SOH, 0x79] + [0x00] * 15 + [EOP])
    glued79 = false79 + bj_79_42
    parts = split_packets(glued79)
    if parts != [bj_79_42]:
        errors.append(
            f"split_packets 0x79 interior EOP crop: {[len(p) for p in parts]!r}"
        )

    inj52 = _pkt(
        "05,52,01,38,D7,00,94,04,08,00,00,00,00,00,00,00,00,"
        "A6,5A,04,00,00,00,00,00,00,00,00,00,AF,86,0D"
    )
    partial52 = inj52[:24]
    norm52 = normalize_window(partial52)
    diag52 = window_skip_diag_fields(partial52, normalized=norm52)
    if diag52.get("opcode_guess") != "52":
        errors.append("window_skip_diag_fields partial 0x52 opcode")
    if diag52.get("wire_len_short_by") != [8]:
        errors.append(f"window_skip_diag_fields partial 0x52 short_by: {diag52!r}")

    master01 = _pkt(
        "05,01,00,01,F7,CE,00,01,F7,CE,8C,00,F3,E7,23,CC,0D"
    )
    if len(master01) != 17:
        errors.append(f"master01 test vector len {len(master01)}")
    ok, reason = eth_forward_ok(master01)
    if not ok:
        errors.append(f"eth_forward_ok master01 0x01 grant: {reason}")
    parts01 = split_packets(master01)
    if parts01 != [master01]:
        errors.append(f"split_packets 0x01 grant: {[len(p) for p in parts01]!r}")

    slave_conf01 = _pkt(
        "05,01,01,D2,49,00,94,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,71,66,0D"
    )
    if len(slave_conf01) != 32:
        errors.append(f"slave_conf01 test vector len {len(slave_conf01)}")
    ok, reason = eth_forward_ok(slave_conf01)
    if not ok:
        errors.append(f"eth_forward_ok slave_conf01 0x01 config req: {reason}")
    parts_conf01 = split_packets(slave_conf01)
    if parts_conf01 != [slave_conf01]:
        errors.append(
            f"split_packets 0x01 config req: {[len(p) for p in parts_conf01]!r}"
        )

    master_conf09 = bytes([SOH, 0x09]) + bytes(1167) + bytes([EOP])
    if len(master_conf09) != 1170:
        errors.append(f"master_conf09 test vector len {len(master_conf09)}")
    ok, reason = eth_forward_ok(master_conf09)
    if not ok:
        errors.append(f"eth_forward_ok master_conf09 0x09 config rsp: {reason}")
    parts_conf09 = split_packets(master_conf09)
    if parts_conf09 != [master_conf09]:
        errors.append(
            f"split_packets 0x09 config rsp: {[len(p) for p in parts_conf09]!r}"
        )
    master_conf09_legacy = bytes([SOH, 0x09]) + bytes(1091) + bytes([EOP])
    if len(master_conf09_legacy) != 1094:
        errors.append(f"master_conf09_legacy test vector len {len(master_conf09_legacy)}")
    parts_conf09_leg = split_packets(master_conf09_legacy)
    if parts_conf09_leg != [master_conf09_legacy]:
        errors.append(
            f"split_packets 0x09 legacy 1094: {[len(p) for p in parts_conf09_leg]!r}"
        )

    grant5a = _pkt(
        "05,5A,01,38,D7,00,94,07,01,00,01,F7,D2,00,01,F7,D2,A8,00,F4,03,65,13,0D"
    )
    if len(grant5a) != 24:
        errors.append(f"grant5a test vector len {len(grant5a)}")
    ok, reason = eth_forward_ok(grant5a)
    if not ok:
        errors.append(f"eth_forward_ok grant5a 0x5A: {reason}")
    parts5a = split_packets(grant5a)
    if parts5a != [grant5a]:
        errors.append(f"split_packets 0x5A grant: {[len(p) for p in parts5a]!r}")

    spinjack72 = _pkt(
        "05,72,01,38,D7,0F,96,05,00,00,00,00,00,00,00,00,00,D0,1A,01,00,00,00,00,"
        "00,00,00,00,00,31,21,0D"
    )
    if len(spinjack72) != 32:
        errors.append(f"spinjack72 test vector len {len(spinjack72)}")
    ok, reason = eth_forward_ok(spinjack72)
    if not ok:
        errors.append(f"eth_forward_ok spinjack72 0x72 claim: {reason}")
    parts72 = split_packets(spinjack72)
    if parts72 != [spinjack72]:
        errors.append(f"split_packets 0x72 claim: {[len(p) for p in parts72]!r}")

    grant7a = bytearray(grant5a)
    grant7a[1] = 0x7A
    grant7a = bytes(grant7a)
    if len(grant7a) != 24:
        errors.append(f"grant7a test vector len {len(grant7a)}")
    ok, reason = eth_forward_ok(grant7a)
    if not ok:
        errors.append(f"eth_forward_ok grant7a 0x7A: {reason}")
    parts7a = split_packets(grant7a)
    if parts7a != [grant7a]:
        errors.append(f"split_packets 0x7A grant: {[len(p) for p in parts7a]!r}")

    status = bytes.fromhex("050D0100000B93007EA45BCCCE0D")
    x41 = bytes.fromhex("05410137BC09930000F61A0D")
    parts = split_packets(x41 + status + status)
    if len(parts) != 3 or len(parts[1]) != 14:
        errors.append(f"split_packets x41+status: {[len(p) for p in parts]!r}")

    # Golden 2045 trunk vectors (interior 0x0D must not crop).
    t6_37 = bytes.fromhex(
        "052001A45B079100040000000000000000E7BB020000000000000000001238040E2430400D"
    )
    if len(t6_37) != 37:
        errors.append(f"t6_37 vector len {len(t6_37)}")
    parts = split_packets(t6_37)
    if len(parts) != 1 or len(parts[0]) != 37:
        errors.append(f"split_packets 0x20 wire len 37: {[len(p) for p in parts]!r}")

    poker_50 = bytes.fromhex(
        "055001A45B009400010000000000000000CEBB0200000000000000000025650D"
    )
    parts = split_packets(poker_50)
    if len(parts) != 1 or len(parts[0]) != 32:
        errors.append(f"split_packets 0x50 wire len 32: {[len(p) for p in parts]!r}")

    lotto_60 = _pkt(
        "05,60,01,A4,5B,08,95,07,01,00,00,00,00,00,00,00,00,83,BB,02,00,00,00,00,"
        "00,00,00,00,00,03,88,41,02,02,27,06,00,93,60,74,00,00,53,30,00,69,03,0D"
    )
    if len(lotto_60) != 48:
        errors.append(f"lotto_60 vector len {len(lotto_60)}")
    parts = split_packets(lotto_60)
    if len(parts) != 1 or len(parts[0]) != 48:
        errors.append(f"split_packets 0x60 wire len 48: {[len(p) for p in parts]!r}")

    bj_79_31 = _pkt(
        "05,79,01,A4,5B,0E,96,00,00,00,00,00,00,00,00,00,00,29,00,6B,8B,37,08,28,"
        "21,36,38,29,0B,05,0D"
    )
    if len(bj_79_31) != 31:
        errors.append(f"bj_79_31 vector len {len(bj_79_31)}")
    parts = split_packets(bj_79_31)
    if len(parts) != 1 or len(parts[0]) != 31:
        errors.append(f"split_packets 0x79 wire len 31: {[len(p) for p in parts]!r}")

    if not uart_window_lead_is_native_wager(0x70):
        errors.append("uart_window_lead_is_native_wager 0x70")
    if not uart_window_lead_is_response_echo(0x79):
        errors.append("uart_window_lead_is_response_echo 0x79")
    if uart_window_lead_is_response_echo(0x70):
        errors.append("uart_window_lead_is_response_echo must reject 0x70")
    if not uart_echo_false_glued_wager(0x70, packets_in_window=2, uart_first_byte=0x79):
        errors.append("uart_echo_false_glued_wager glued pc70")
    if uart_echo_false_glued_wager(0x71, packets_in_window=2, uart_first_byte=0x79):
        errors.append("uart_echo_false_glued_wager must allow glued pc71")
    if uart_echo_false_glued_wager(0x70, packets_in_window=1, uart_first_byte=0x79):
        errors.append("uart_echo_false_glued_wager solo must not block pc70 (echo_wager handles)")

    if len(bj_79_42) != 42:
        errors.append(f"bj_79_42 vector len {len(bj_79_42)}")
    parts = split_packets(bj_79_42)
    if len(parts) != 1 or len(parts[0]) != 42:
        errors.append(f"split_packets 0x79 wire len 42: {[len(p) for p in parts]!r}")

    # 1835 wire-len census (AF1D trunk).
    bingo_30 = _pkt(
        "05,30,01,1D,AF,06,92,00,10,00,00,00,00,00,00,00,00,F2,C1,00,00,00,00,00,"
        "00,00,00,00,00,48,4C,24,43,C9,00,54,24,0E,12,C0,68,20,47,08,4C,22,64,"
        "46,21,A0,0B,1A,0C,2C,20,06,4A,0D"
    )
    if len(bingo_30) != 58:
        errors.append(f"bingo_30 vector len {len(bingo_30)}")
    parts = split_packets(bingo_30)
    if len(parts) != 1 or len(parts[0]) != 58:
        errors.append(f"split_packets 0x30 wire len 58: {[len(p) for p in parts]!r}")
    if bingo_play_card_count_from_wager(bingo_30) != 2:
        errors.append(f"bingo_30 play card count {bingo_play_card_count_from_wager(bingo_30)} != 2")
    cards58 = bingo_cards_from_wager(bingo_30)
    if len(cards58) != 2:
        errors.append(f"bingo_cards_from_wager 58 B count {len(cards58)} != 2")

    def _bingo_wager_play_prefix(src: bytes, play_cards: int) -> bytes:
        wl = bingo_wager_wire_len_for_play_count(play_cards)
        frame = bytearray(src[: wl - 3])
        frame.extend((0, 0, EOP))
        pog422_set_frame_checksum(frame)
        return bytes(frame)

    # 2026-07-04 bench — pos2 TID 38D7, game 9206, 4-card wager (80 B fixed frame).
    bingo_30_80 = _pkt(
        "05,30,01,38,D7,06,92,00,10,00,00,00,00,00,00,00,00,22,DC,0B,00,00,00,00,00,"
        "00,00,00,00,00,54,0C,99,01,83,10,14,2C,AB,00,10,4D,14,2A,AA,00,82,32,44,45,"
        "19,0C,21,23,A2,40,18,19,05,70,0E,41,D2,08,00,1D,4A,0A,2C,06,00,00,00,00,00,"
        "00,00,68,B0,0D"
    )
    if len(bingo_30_80) != 80:
        errors.append(f"bingo_30_80 vector len {len(bingo_30_80)}")
    parts = split_packets(bingo_30_80)
    if len(parts) != 1 or len(parts[0]) != 80:
        errors.append(f"split_packets 0x30 wire len 80: {[len(p) for p in parts]!r}")
    if not pog422_frame_checksum_ok(bingo_30_80):
        errors.append("bingo_30_80 checksum invalid")
    raw_blocks = bingo_card_raw_blocks_from_wager(bingo_30_80)
    if len(raw_blocks) != 4:
        errors.append(f"bingo_card_raw_blocks count {len(raw_blocks)} != 4")
    if bingo_play_card_count_from_wager(bingo_30_80) != 4:
        errors.append(f"bingo_30_80 play slot count {bingo_play_card_count_from_wager(bingo_30_80)} != 4")
    if bingo_wager_byte8_play_credits(bingo_30_80) != 0x10:
        errors.append("bingo_30_80 byte8 play credits != 0x10")
    overlap = bingo_operator_reference_overlap_score(bingo_30_80)
    if overlap >= 48:
        errors.append(
            f"bingo legacy codec unexpectedly matches operator refs ({overlap}/96)"
        )
    for card_idx, grid in enumerate(BINGO_OPERATOR_REFERENCE_CARDS):
        enc = bingo_grid_to_card_bytes_legacy_cgs(grid)
        back = bingo_card_bytes_to_grid_legacy_cgs(enc)
        if not bingo_grids_equal(back, grid):
            errors.append(f"bingo legacy codec card {card_idx + 1} roundtrip failed")

    bingo_39 = _pkt(
        "05,39,01,1D,AF,06,92,00,00,00,06,1A,85,00,00,00,00,1B,00,6C,94,36,09,2D,"
        "4B,45,3A,01,25,05,20,12,18,1E,39,17,0B,3F,33,13,04,26,42,2F,28,37,38,08,"
        "29,3D,2B,47,21,0F,22,14,41,30,3C,03,48,1F,2E,0A,1A,1B,46,35,43,15,1C,76,"
        "1E,0D"
    )
    if len(bingo_39) != 74:
        errors.append(f"bingo_39 vector len {len(bingo_39)}")
    parts = split_packets(bingo_39)
    if len(parts) != 1 or len(parts[0]) != 74:
        errors.append(f"split_packets 0x39 wire len 74: {[len(p) for p in parts]!r}")

    # 1859 card-matrix pos1 packets.raw line 36 (4 cards, 48 B).
    t6_48 = _pkt(
        "05,20,01,1D,AF,07,91,06,10,00,00,00,00,00,00,00,00,D2,BE,00,00,00,00,00,"
        "00,00,00,00,00,01,31,41,0E,C3,40,00,0F,9A,50,00,0E,13,05,08,0E,B3,D8,0D"
    )
    if len(t6_48) != 48:
        errors.append(f"t6_48 vector len {len(t6_48)}")
    parts = split_packets(t6_48)
    if len(parts) != 1 or len(parts[0]) != 48:
        errors.append(f"split_packets 0x20 wire len 48: {[len(p) for p in parts]!r}")

    bl: set[str] = set()
    bl_order: list[str] = []
    if not eth_inject_tid_blacklist_add(bl, bl_order, "38D7"):
        errors.append("eth_inject_tid_blacklist_add first")
    if eth_inject_tid_blacklist_add(bl, bl_order, "38d7"):
        errors.append("eth_inject_tid_blacklist_add case-insensitive dedupe")
    if local_slave_tid_uart_learn_ok("38D7", eth_inject_tid_blacklist=bl):
        errors.append("local_slave_tid_uart_learn_ok must reject eth-inject blacklist")
    if not local_slave_tid_uart_learn_ok("37BC", eth_inject_tid_blacklist=bl):
        errors.append("local_slave_tid_uart_learn_ok must allow non-blacklisted TID")
    if not local_slave_tid_uart_learn_ok("37BC", eth_inject_tid_blacklist=bl):
        errors.append("local_slave_tid_uart_learn_ok must allow non-blacklisted TID")
    poker_wager = bytes([0x05, 0x50, 0x01, 0x37, 0xBC, 0x03, 0x94, 0x01, 0x00, 0x0D])
    if parse_packet_meta(poker_wager).tid_hex != "37BC":
        errors.append("terminal tid wire order (37BC)")
    for i in range(ETH_INJECT_TID_BLACKLIST_MAX + 5):
        eth_inject_tid_blacklist_add(bl, bl_order, f"{i:04X}")
    if len(bl) != ETH_INJECT_TID_BLACKLIST_MAX:
        errors.append(
            f"eth_inject_tid_blacklist cap {len(bl)} != {ETH_INJECT_TID_BLACKLIST_MAX}"
        )
    if "38D7" in bl or "38D7" in bl_order:
        errors.append("eth_inject_tid_blacklist FIFO must evict 38D7")

    # Bank bell fanout: winner-TID x9 with progwinval>0 must be treated bank-wide.
    jpot_x9 = bytearray(20)
    jpot_x9[0] = SOH
    jpot_x9[1] = KENO_RESULT_MSG
    jpot_x9[3] = 0x37
    jpot_x9[4] = 0xBC
    jpot_x9[-1] = EOP
    if is_bank_progressive_jackpot_frame(bytes(jpot_x9)):
        errors.append("is_bank_progressive_jackpot_frame must reject progwin=0")
    jpot_x9[PROGRESSIVE_PROGWIN_OFFSET : PROGRESSIVE_PROGWIN_OFFSET + 4] = (
        500_088
    ).to_bytes(4, "big")
    if not is_bank_progressive_jackpot_frame(bytes(jpot_x9)):
        errors.append("is_bank_progressive_jackpot_frame must accept progwin>0 x9")
    wager_with_progwin = bytearray(jpot_x9)
    wager_with_progwin[1] = 0x40  # keno wager (x0) — request, not an outcome
    if is_bank_progressive_jackpot_frame(bytes(wager_with_progwin)):
        errors.append("is_bank_progressive_jackpot_frame must reject non-x9")
    # Skill grants answer a slave claim and carry the amount (poker royal, SpinJack).
    for grant_msg in sorted(SKILL_JPOT_GRANT_MSG_TYPES):
        grant = bytearray(jpot_x9)
        grant[1] = grant_msg
        if not is_bank_progressive_jackpot_frame(bytes(grant)):
            errors.append(
                f"is_bank_progressive_jackpot_frame must accept grant {grant_msg:02X}"
            )
    for claim_msg in (0x52, 0x72):
        claim = bytearray(jpot_x9)
        claim[1] = claim_msg
        if is_bank_progressive_jackpot_frame(bytes(claim)):
            errors.append(
                f"is_bank_progressive_jackpot_frame must reject claim {claim_msg:02X}"
            )
    if is_bank_progressive_jackpot_frame(None) or is_bank_progressive_jackpot_frame(b""):
        errors.append("is_bank_progressive_jackpot_frame must reject empty")

    # Bank jpot x1 priority class (EG 3.0.1.24)
    armed: dict[str, int] = {"37BC": time.perf_counter_ns() + 1_000_000_000}
    if not is_bank_jpot_x1_priority_ack(0x81, "37BC", armed):
        errors.append("is_bank_jpot_x1_priority_ack must accept armed TID x1")
    if is_bank_jpot_x1_priority_ack(0x81, "1DAF", armed):
        errors.append("is_bank_jpot_x1_priority_ack must reject unarmed TID")
    if is_bank_jpot_x1_priority_ack(0x89, "37BC", armed):
        errors.append("is_bank_jpot_x1_priority_ack must reject non-ack")
    if bank_progressive_jackpot_winner_tid(bytes(jpot_x9)) != "37BC":
        errors.append("bank_progressive_jackpot_winner_tid")
    if inject_priority_class(
        0x81,
        tid_hex="37BC",
        bank_jpot_x1_armed_until_ns=armed,
    ) != INJECT_PRIORITY_BANK_JPOT_X1:
        errors.append("inject_priority_class bank jpot x1")
    if inject_priority_class(SCR_MSG) != INJECT_PRIORITY_SCR:
        errors.append("inject_priority_class SCR must beat soft STOM")
    if inject_priority_class(SCR_REPLY) != INJECT_PRIORITY_SCR:
        errors.append("inject_priority_class SCR_REPLY")
    if inject_priority_class(0x0A) != INJECT_PRIORITY_STOM:
        errors.append("inject_priority_class 0A stays STOM")
    if inject_priority_class(0x02) != INJECT_PRIORITY_STOM:
        errors.append("inject_priority_class stom")
    if inject_priority_class(0x80) != INJECT_PRIORITY_NORMAL:
        errors.append("inject_priority_class normal")
    # EG 3.0.1.33: any TID's x1 / 0D is a close leg (never bank-jpot defer).
    if not is_gameplay_close_leg(0x81):
        errors.append("is_gameplay_close_leg x1")
    if not is_gameplay_close_leg(0x0D):
        errors.append("is_gameplay_close_leg 0d")
    if not is_gameplay_close_leg(None, source="status_0d_synth"):
        errors.append("is_gameplay_close_leg status_0d_synth")
    if is_gameplay_close_leg(0x49):
        errors.append("is_gameplay_close_leg must reject x9")
    if is_gameplay_close_leg(0x40):
        errors.append("is_gameplay_close_leg must reject x0")

    if bytes_hex_line(bytes([0x01, 0xAB, 0x0D])) != "01,AB,0D":
        errors.append("bytes_hex_line")
    if not channel_uses_raw_hs_window(BUS_CHANNELS["ois"]):
        errors.append("channel_uses_raw_hs_window ois")
    if channel_uses_raw_hs_window(BUS_CHANNELS["machine"]):
        errors.append("channel_uses_raw_hs_window machine must be false")
    ok, _ = ois_raw_forward_ok(bytes([0x01, 0xAB]))
    if not ok:
        errors.append("ois_raw_forward_ok min len")
    ok, reason = ois_raw_forward_ok(b"")
    if ok or reason != "empty":
        errors.append("ois_raw_forward_ok empty")
    tagged = fixture_eth_tag_payload(BUS_CHANNELS["machine"], b"\x05\x03", dual=True)
    if tagged != bytes([FIXTURE_ETH_TAG_MACHINE, 0x05, 0x03]):
        errors.append("fixture_eth_tag_payload machine")
    ch, body, skip = fixture_eth_untag_payload(
        bytes([FIXTURE_ETH_TAG_OIS, 0x01, 0xAB]), dual=True
    )
    if ch != "ois" or body != b"\x01\xAB" or skip:
        errors.append("fixture_eth_untag_payload ois")
    ch2, body2, skip2 = fixture_eth_untag_payload(b"\x05", dual=True)
    if ch2 is not None or skip2 != "dual_tag_missing":
        errors.append("fixture_eth_untag_payload missing tag")
    ch3, body3, skip3 = fixture_eth_untag_payload(b"\x05\x0D", dual=True)
    if ch3 is not None or body3 != b"\x05\x0D" or skip3 != "dual_tag_unknown":
        errors.append("fixture_eth_untag_payload unknown tag")
    both = resolve_bridge_channels("both")
    if len(both) != 2 or both[0].key != "machine":
        errors.append("resolve_bridge_channels both")

    with tempfile.TemporaryDirectory() as td:
        out = Path(td)
        ep = out / EVENTS_JSONL
        pp = out / PACKETS_RAW
        ep.write_text('{"ev":"test"}\n' * 200, encoding="utf-8")
        pp.write_text("05,79\n" * 100, encoding="utf-8")
        meta = rotate_debug_capture_logs(
            out,
            max_bytes=500,
            utc_daily=False,
            utc_day="2026-07-03",
        )
        if meta is None or meta.get("reason") != "size":
            errors.append("rotate_debug_capture_logs size trigger")
        elif ep.is_file():
            errors.append("rotate_debug_capture_logs active events not renamed")
        elif not list(out.glob(f"{EVENTS_JSONL}.*")):
            errors.append("rotate_debug_capture_logs events archive missing")
        elif not list(out.glob(f"{PACKETS_RAW}.*")):
            errors.append("rotate_debug_capture_logs packets archive missing")
        meta2 = rotate_debug_capture_logs(
            out,
            max_bytes=500,
            utc_daily=True,
            utc_day="2026-07-02",
        )
        if meta2 is not None:
            errors.append("rotate_debug_capture_logs empty active should not rotate")

        # packets.raw must keep complete HS copper even when split_packets fails.
        identity = FixtureIdentity(fixture_pos=2, pi_label="selftest", role="listen")
        elog = EventLog(out, identity, echo=False, prod_mode=False, hs_witness=False)
        try:
            short_scr = bytearray(
                bytes([0x05, 0x03, 0x01]) + bytes(16) + bytes([0x0D])
            )  # 20 B — not a full 32 B SCR
            assert len(short_scr) == 20
            elog.finalize_handshake_window(
                short_scr,
                window_start_perf_ns=1,
                window_start_mono_ns=1,
                first_byte_perf_ns=1,
                last_byte_perf_ns=2,
                uart_first_byte=0x05,
            )
            elog.close()
            lines = [
                ln.strip()
                for ln in (out / PACKETS_RAW).read_text(encoding="utf-8").splitlines()
                if ln.strip()
            ]
            want = bytes_hex_line(bytes(short_scr))
            if want not in lines:
                errors.append(
                    f"packets.raw must keep full skip window hex; got {lines!r}"
                )
            ev_lines = (out / EVENTS_JSONL).read_text(encoding="utf-8").splitlines()
            skip_rows = [
                json.loads(ln)
                for ln in ev_lines
                if ln.strip() and '"window_skip"' in ln
            ]
            if not skip_rows or skip_rows[-1].get("hs_window_raw") != want:
                errors.append("window_skip must include hs_window_raw matching packets.raw")
        except Exception as exc:  # noqa: BLE001 — self-test collects errors
            errors.append(f"packets.raw full-hs self-test: {exc!r}")
            try:
                elog.close()
            except Exception:
                pass

    return errors

UART_BAUD = 57600
XR_USB_SERIAL_GET_REG = 0xC0047601
XR_USB_SERIAL_SET_REG = 0xC0047602
XR_USB_SERIAL_SET_GPIO7 = 0x4004760D

GPIO_MODE_REG = 0x0C
GPIO_DIR_REG = 0x0D
GPIO_STAT_REG = 0x10
# Match inject-shim pog422_inject.c manual SLV (ch2 / ttyXRUSB1): MODE 0x0300, DIR 0xA8.
# HandshakeMonitor used to write MODE=0 + DIR=0x80 on every open/inject — that wiped
# driver 1N manual GPIO7 so set_tx() toggled HS without a proper DE data window.
POG422_GPIO_MODE_MANUAL_SLV = 0x0300
POG422_GPIO_MODE_RS485_AUTO = 0x0339
POG422_GPIO7_DIR = 0xA8  # DTR + RTS + GPIO7 outputs (0x28 | 0x80)
GPIO7_DIR_MASK = 0x80  # GPIO7 bit only (legacy alias)
GPIO1_STAT_MASK = 0x02

ETH_TYPE = 0x88B5
ETH_P_ALL = 0x0003

# x0 request -> x9 response (high nibble + 9)
REQUEST_TO_RESPONSE: dict[int, int] = {
    0x20: 0x29,
    0x30: 0x39,
    0x40: 0x49,
    0x50: 0x59,
    0x60: 0x69,
    0x70: 0x79,
    0x80: 0x89,
}

GAMEPLAY_ACK_MSG_TYPES = frozenset({0x21, 0x31, 0x41, 0x51, 0x61, 0x71, 0x81})

GAMEPLAY_REQUEST_MSG_TYPES = frozenset(REQUEST_TO_RESPONSE.keys())

GAMEPLAY_RESPONSE_MSG_TYPES = frozenset(set(REQUEST_TO_RESPONSE.values()))

# Outcome (x9) byte 17 holds the Master close counter for this round on copper captures.
STATUS_0D_X9_COUNTER_BYTE = 17

# Default master progressive entity tail (0x0A bytes 9-10) when not yet seen this session.
STATUS_0D_DEFAULT_BANK_TID_TAIL = bytes((0xA4, 0x5B))

# Lab-only Keno progressive outcome synth (TouchEasy Keno bus id 9309, msg 0x49).
KENO_RESULT_MSG = 0x49
TOUCHEASY_KENO_GAME_ID = "9309"
KENO_RESULT_ACTIVEWINS_OFFSET = 8
KENO_RESULT_PROGWIN_OFFSET = 13
# 20 draw balls at bytes 21–40 (44-byte 0x49), on-screen values 1–80 raw.
# Byte 17 = bank close counter (preserve). Stable layout — same for natural Master
# and lab synth (do not shift). Old lab path wrongly used offset 19 + wire−2.
KENO_RESULT_BALL_OFFSET = 21
KENO_LAB_DEFAULT_DISPLAY_BALLS = tuple(range(1, 21))

# Progressive block on x9 opcodes (1995 SLAVE.C stripjpotdata — BE32 cents).
PROGRESSIVE_ACTIVEWINS_OFFSET = 8
PROGRESSIVE_PROGDSP_OFFSET = 9
PROGRESSIVE_PROGWIN_OFFSET = 13


def u32_be_at(frame: bytes, offset: int) -> int:
    if len(frame) < offset + 4:
        return 0
    return int.from_bytes(frame[offset : offset + 4], "big")


def u32_le_at(frame: bytes, offset: int) -> int:
    if len(frame) < offset + 4:
        return 0
    return int.from_bytes(frame[offset : offset + 4], "little")


def progressive_cents_to_dollars(cents: int) -> float:
    return int(cents) / 100.0


def progressive_progwin_cents_be(frame: bytes, offset: int = PROGRESSIVE_PROGWIN_OFFSET) -> int:
    """Jackpot win amount in cents (1995 slave: BE32 at bytes 13–16)."""
    return u32_be_at(frame, offset)


def progressive_progdsp_cents_be(frame: bytes, offset: int = PROGRESSIVE_PROGDSP_OFFSET) -> int:
    """Progressive meter display in cents (BE32 at bytes 9–12)."""
    return u32_be_at(frame, offset)


def resolve_lab_progwin_cents(frame: bytes, override: int | None = None) -> int:
    """Resolve lab progwinval in cents for synth (override > wire progwin BE > meter BE)."""
    if override is not None:
        return int(override)
    win = progressive_progwin_cents_be(frame)
    if win:
        return win
    meter = progressive_progdsp_cents_be(frame)
    return meter or 1


def keno_wire_byte_for_display(display_ball: int) -> int:
    """Deprecated: T340 Keno balls are raw 1–80 on wire (no −2 encoding).

    Kept for older capture tooling; lab synth writes display values directly.
    """
    return int(display_ball) & 0xFF


def keno_display_ball_from_wire(wire_byte: int) -> int:
    """Deprecated identity: wire byte is already the on-screen ball 1–80."""
    return int(wire_byte) & 0xFF


def keno_lab_default_wire_balls() -> tuple[int, ...]:
    """Lab default draw: display balls 1..20 written raw on wire."""
    return KENO_LAB_DEFAULT_DISPLAY_BALLS


def keno_result_wire_lens() -> tuple[int, ...]:
    return wire_lengths_for_cmd(KENO_RESULT_MSG)


def build_keno_jpot_synth(
    template: bytes,
    *,
    balls: tuple[int, ...] | list[int] | None = None,
    jpot_amount: int | None = None,
) -> bytes | None:
    """Build lab Keno x9 (0x49) with forced jackpot progwinval and draw balls.

    ``balls`` are **on-screen ball numbers** (1–80), default 1..20 (10/10 on spots 1–10).
    Wire encoding: raw display value at ``KENO_RESULT_BALL_OFFSET`` (21).

    NEVER use in production — corrupts Master/slave progressive accounting if abused.
    """
    if len(template) not in keno_result_wire_lens():
        return None
    if len(template) < KENO_RESULT_BALL_OFFSET + 20:
        return None
    frame = bytearray(template)
    frame[KENO_RESULT_ACTIVEWINS_OFFSET] = 1
    amount = resolve_lab_progwin_cents(template, jpot_amount)
    be4 = amount.to_bytes(4, "big")
    frame[PROGRESSIVE_PROGDSP_OFFSET : PROGRESSIVE_PROGDSP_OFFSET + 4] = be4
    frame[KENO_RESULT_PROGWIN_OFFSET : KENO_RESULT_PROGWIN_OFFSET + 4] = be4
    display_balls = tuple(balls) if balls is not None else KENO_LAB_DEFAULT_DISPLAY_BALLS
    if len(display_balls) != 20:
        return None
    for i, display_ball in enumerate(display_balls):
        if not (1 <= display_ball <= 80):
            return None
        frame[KENO_RESULT_BALL_OFFSET + i] = int(display_ball) & 0xFF
    pog422_set_frame_checksum(frame)
    return bytes(frame)


# Lab-only Touch 6 Lotto outcome synth (bus id 9107, wire bytes 07,91, msg 0x29).
TOUCH6_RESULT_MSG = 0x29
TOUCH6_GAME_ID = "9107"
TOUCH6_LOT6_PLAY_MASK = 0x02000000
TOUCH6_WAGER_CARD_OFFSET = 29
TOUCH6_CARD_COUNT = 4
# T340 30-byte 0x29: draw balls @ bytes 21–26 (bench 2026-07-03). Was 19 — same +2 class
# as Keno (was 19, correct 17): partial overwrite left native tail on wire → UI showed 3,4,5,6,24,2
# while lab counted jpot vs forced 1–6 at the wrong offset.
TOUCH6_RESULT_BALL_OFFSET = 21
TOUCH6_LAB_DEFAULT_BALLS = (1, 2, 3, 4, 5, 6)


def touch6_result_wire_lens() -> tuple[int, ...]:
    return wire_lengths_for_cmd(TOUCH6_RESULT_MSG)


def touch6_card_picks_from_wager(wager: bytes) -> tuple[int, ...]:
    """Four LE card masks from 0x20 @ bytes 29–44 (1995 SLAVE.C formslavhead lot6)."""
    return tuple(
        u32_le_at(wager, TOUCH6_WAGER_CARD_OFFSET + c * 4) for c in range(TOUCH6_CARD_COUNT)
    )


def touch6_ball_draw_flags(balls: tuple[int, ...]) -> int:
    flags = 0
    for ball in balls:
        if not (1 <= ball <= 25):
            return 0
        flags |= 1 << (ball - 1)
    return flags


def touch6_count_jpot_cards(wager: bytes | None, balls: tuple[int, ...]) -> int:
    """Count cards with PLAY set and 6/6 hits vs drawn balls (MASTER.C request_lot6)."""
    if wager is None or len(wager) < TOUCH6_WAGER_CARD_OFFSET + TOUCH6_CARD_COUNT * 4:
        return 0
    ball_flags = touch6_ball_draw_flags(balls)
    if not ball_flags:
        return 0
    count = 0
    for card in touch6_card_picks_from_wager(wager):
        if not (card & TOUCH6_LOT6_PLAY_MASK):
            continue
        picks = card & 0x01FFFFFF
        hits = 0
        for j in range(25):
            if (1 << j) & ball_flags & picks:
                hits += 1
        if hits == 6:
            count += 1
    return count


def build_touch6_result_synth(
    template: bytes,
    *,
    balls: tuple[int, ...] | list[int] | None = None,
    jpot_amount: int | None = None,
    wager: bytes | None = None,
) -> bytes | None:
    """Build lab Touch 6 x9 (0x29): forced draw balls + conditional progressive block.

    Draw balls are on-screen values 1–25 at bytes 21–26 (not Keno wire−2 encoding).
    ``activewins`` and ``progwinval`` are set only when the remembered 0x20 wager
    has one or more PLAY cards with 6/6 vs the forced draw (bench 3-game sequence).

    NEVER use in production.
    """
    if len(template) not in touch6_result_wire_lens():
        return None
    if len(template) < TOUCH6_RESULT_BALL_OFFSET + 6:
        return None
    draw = tuple(balls) if balls is not None else TOUCH6_LAB_DEFAULT_BALLS
    if len(draw) != 6:
        return None
    if touch6_ball_draw_flags(draw) == 0:
        return None
    frame = bytearray(template)
    for i, ball in enumerate(draw):
        frame[TOUCH6_RESULT_BALL_OFFSET + i] = ball & 0xFF
    jpot_cards = touch6_count_jpot_cards(wager, draw)
    if jpot_cards > 0:
        frame[PROGRESSIVE_ACTIVEWINS_OFFSET] = jpot_cards & 0xFF
        amount = resolve_lab_progwin_cents(template, jpot_amount)
        be4 = amount.to_bytes(4, "big")
        frame[PROGRESSIVE_PROGDSP_OFFSET : PROGRESSIVE_PROGDSP_OFFSET + 4] = be4
        frame[PROGRESSIVE_PROGWIN_OFFSET : PROGRESSIVE_PROGWIN_OFFSET + 4] = be4
    else:
        frame[PROGRESSIVE_ACTIVEWINS_OFFSET] = 0
        frame[PROGRESSIVE_PROGWIN_OFFSET : PROGRESSIVE_PROGWIN_OFFSET + 4] = b"\0\0\0\0"
    pog422_set_frame_checksum(frame)
    return bytes(frame)


# Lab-only SuperPick outcome synth (POG_SUPR bus id 9507, wire bytes 07,95, msg 0x69).
SUPERPICK_WAGER_MSG = 0x60
SUPERPICK_RESULT_MSG = 0x69
SUPERPICK_GAME_ID = "9508"
SUPERPICK_PLAY_MASK = 0x02000000
SUPERPICK_WAGER_CARD_OFFSET = 29
SUPERPICK_CARD_COUNT = 4
# Same 30-byte x9 layout as Touch 6 on T340 (bench 2026-07-03).
SUPERPICK_RESULT_BALL_OFFSET = 21
SUPERPICK_LAB_DEFAULT_BALLS = (0, 0, 0, 0, 0, 0)


def superpick_result_wire_lens() -> tuple[int, ...]:
    return wire_lengths_for_cmd(SUPERPICK_RESULT_MSG)


def superpick_card_picks_from_wager(wager: bytes) -> tuple[int, ...]:
    """Four LE card masks from 0x60 @ bytes 29–44 (1995 MASTER.C stom_supr)."""
    return tuple(
        u32_le_at(wager, SUPERPICK_WAGER_CARD_OFFSET + c * 4)
        for c in range(SUPERPICK_CARD_COUNT)
    )


def superpick_ball_draw_flags(balls: tuple[int, ...]) -> int:
    flags = 0
    for ball in balls:
        if not (0 <= ball <= 9):
            return 0
        flags |= 1 << ball
    return flags


def superpick_nibble_at(card: int, place: int) -> int:
    return (card >> (place * 4)) & 0x0F


def superpick_first_hits(card: int, balls: tuple[int, ...]) -> int:
    """Consecutive matches from position 0 leftward (SUPER.C first[])."""
    if len(balls) != 6:
        return 0
    first = 0
    for j in range(6):
        if balls[j] == superpick_nibble_at(card, j):
            if j == first:
                first += 1
        else:
            break
    return first


def superpick_any_hits(card: int, balls: tuple[int, ...]) -> int:
    """Matches anywhere in the six pick positions (SUPER.C any[])."""
    if len(balls) != 6:
        return 0
    return sum(1 for j in range(6) if balls[j] == superpick_nibble_at(card, j))


def superpick_card_jpot_eligible(card: int, balls: tuple[int, ...]) -> bool:
    """Progressive: max-bet card with FIRST 5 or ALL 6 (first_hits >= 5).

    Paytable also pays ANY-column hits separately; those are regular wins only.
    Bench assumes max bet when PLAY is set (bet not decoded from 0x60 byte 8).

    Note: 1995 MASTER.C ``request_supr()`` k-loop (6th match resets k) implemented a
    Foxwoods-only config where progressive was **5-of-6 exclusively** — not the
    default SuperPick rule. Lab synth follows standard FIRST≥5 (FIRST 5 or ALL 6).
    """
    if not (card & SUPERPICK_PLAY_MASK):
        return False
    first = superpick_first_hits(card, balls)
    return first >= 5


def superpick_count_jpot_cards(wager: bytes | None, balls: tuple[int, ...]) -> int:
    """Count PLAY cards eligible for progressive (FIRST 5 or ALL 6 vs forced draw)."""
    if wager is None or len(wager) < SUPERPICK_WAGER_CARD_OFFSET + SUPERPICK_CARD_COUNT * 4:
        return 0
    if superpick_ball_draw_flags(balls) == 0:
        return 0
    count = 0
    for card in superpick_card_picks_from_wager(wager):
        if superpick_card_jpot_eligible(card, balls):
            count += 1
    return count


def build_superpick_result_synth(
    template: bytes,
    *,
    balls: tuple[int, ...] | list[int] | None = None,
    jpot_amount: int | None = None,
    wager: bytes | None = None,
) -> bytes | None:
    """Build lab SuperPick x9 (0x69): forced draw + conditional progressive block.

    Draw balls are on-screen values 0–9 at bytes 21–26. Progressive ``activewins``
    / ``progwinval`` when PLAY cards have **FIRST 5 or ALL 6** vs the forced draw
    (``first_hits >= 5``; matches ``SUPER.C`` paytable / operator rules).

    NEVER use in production.
    """
    if len(template) not in superpick_result_wire_lens():
        return None
    if len(template) < SUPERPICK_RESULT_BALL_OFFSET + 6:
        return None
    draw = tuple(balls) if balls is not None else SUPERPICK_LAB_DEFAULT_BALLS
    if len(draw) != 6:
        return None
    if superpick_ball_draw_flags(draw) == 0:
        return None
    frame = bytearray(template)
    for i, ball in enumerate(draw):
        frame[SUPERPICK_RESULT_BALL_OFFSET + i] = ball & 0xFF
    jpot_cards = superpick_count_jpot_cards(wager, draw)
    if jpot_cards > 0:
        frame[PROGRESSIVE_ACTIVEWINS_OFFSET] = jpot_cards & 0xFF
        amount = resolve_lab_progwin_cents(template, jpot_amount)
        be4 = amount.to_bytes(4, "big")
        frame[PROGRESSIVE_PROGDSP_OFFSET : PROGRESSIVE_PROGDSP_OFFSET + 4] = be4
        frame[PROGRESSIVE_PROGWIN_OFFSET : PROGRESSIVE_PROGWIN_OFFSET + 4] = be4
    else:
        frame[PROGRESSIVE_ACTIVEWINS_OFFSET] = 0
        frame[PROGRESSIVE_PROGWIN_OFFSET : PROGRESSIVE_PROGWIN_OFFSET + 4] = b"\0\0\0\0"
    pog422_set_frame_checksum(frame)
    return bytes(frame)


# SuperGold Bingo (POG_BING bus id 9206, wager 0x30 / result 0x39 / ack 0x31).
BINGO_WAGER_MSG = 0x30
BINGO_RESULT_MSG = 0x39
BINGO_ACK_MSG = 0x31
BINGO_GAME_ID = "9206"
BINGO_WAGER_CARD_OFFSET = 29
BINGO_CARD_BYTE_LEN = 12
BINGO_CARD_COUNT = 4
BINGO_PLAY_MASK = 0x02000000
BINGO_CENTER_ROW = 2
BINGO_CENTER_COL = 2
BINGO_RESULT_CALL_OFFSET = 21  # T340 74 B wire — 50 ball values @ 1–75 (bench 2026-07-04)
BINGO_RESULT_CALL_COUNT = 50
BINGO_WAGER_WIRE_LEN = 80  # T340+ fixed four-slot frame (all PLAY counts; bench 2026-07-04)
BINGO_WAGER_PREAMBLE_LEN = 29  # SOH .. byte before card payload (5.52U 80 B wire)
BINGO_WAGER_PAYLOAD_LEN = 48  # 4 × 12 B card slots @ 29..76 — encoding TBD
BINGO_WAGER_TAIL_LEN = 3  # CRC16-BE (2 B) + EOP
BINGO_WAGER_CENSUS_WIRE_LEN = 58  # 1835 AF1D trunk — two sequential 12 B cards
# Legacy SLAVE.C uses 0xFF for OFF slots; T340+ bench uses zero-fill within the 80 B frame.
BINGO_CARD_OFF_LEGACY = 0xFF

# Card byte→grid algorithm is **unknown** on T340+ / POG 5.52U SuperGold (9206).
# Repo Legacy Code/BINGO.C (savebingcard) is pre-5.52U and does **not** match 5.52U
# wire captures — do not treat bench failure as stale data. See validate_bingo_wager_cards.py
# and sessions/2026-07-04_0251-UTC_bingo-4-1-2-3-card-pos2/.
#
# Wire packing (all TBD on 5.52U — separate from grid layout):
#   • byte order within each 12 B slot (forward vs reversed)
#   • nibble order within each byte (high nibble first vs low nibble first)
#   • spot stream order (row-major 0–24 vs column-major, skip center 12)
# Legacy CGS hypothesis uses row-major spots + 2 nibbles/byte (low nibble first).
BINGO_CARD_CODEC_STATUS = "unknown_552u"

# Operator bench reference (2026-07-04 swap-cards session) — four 5×5 grids from
# SUPERGOLD BINGO screenshots (top-left card → bottom-right). **Row-major:**
# grid[row][col] with col 0=B … 4=O; center (2,2) is None (free). Same payload on
# both wagers in archived pos2 capture; legacy codec does NOT match wire @ offset 29.
BingoGrid = tuple[tuple[int | None, ...], ...]
BINGO_OPERATOR_REFERENCE_CARDS: tuple[BingoGrid, ...] = (
    (
        (4, 21, 34, 52, 73),
        (5, 17, 44, 49, 66),
        (8, 19, None, 54, 61),
        (3, 26, 33, 56, 62),
        (13, 22, 39, 48, 65),
    ),
    (
        (3, 17, 44, 56, 61),
        (12, 25, 36, 53, 67),
        (2, 29, None, 55, 72),
        (6, 27, 39, 48, 68),
        (4, 19, 45, 59, 75),
    ),
    (
        (14, 19, 34, 50, 75),
        (7, 22, 31, 59, 66),
        (6, 16, None, 47, 64),
        (1, 26, 33, 52, 68),
        (3, 18, 40, 58, 73),
    ),
    (
        (4, 17, 41, 55, 63),
        (2, 21, 42, 48, 66),
        (8, 26, None, 53, 73),
        (3, 25, 45, 50, 68),
        (9, 27, 31, 56, 70),
    ),
)


def bingo_wager_wire_lens() -> tuple[int, ...]:
    return wire_lengths_for_cmd(BINGO_WAGER_MSG)


def bingo_wager_frame_layers(wager: bytes) -> dict[str, bytes | tuple[bytes, ...]]:
    """Split 80 B SuperGold Bingo wager (0x30) into known preamble, raw payload, tail.

    Does not decode card grids — payload is opaque until 5.52U packing is solved.
    """
    if len(wager) != BINGO_WAGER_WIRE_LEN:
        raise ValueError(f"bingo wager len {len(wager)} != {BINGO_WAGER_WIRE_LEN}")
    pre_end = BINGO_WAGER_PREAMBLE_LEN
    pay_end = pre_end + BINGO_WAGER_PAYLOAD_LEN
    payload = wager[pre_end:pay_end]
    return {
        "preamble": wager[:pre_end],
        "payload": payload,
        "payload_slots": tuple(
            payload[i * BINGO_CARD_BYTE_LEN : (i + 1) * BINGO_CARD_BYTE_LEN]
            for i in range(BINGO_CARD_COUNT)
        ),
        "tail_crc": wager[pay_end : pay_end + 2],
        "tail_eop": wager[pay_end + 2 : pay_end + 3],
    }


def bingo_wager_preamble_fields(preamble: bytes) -> dict[str, int | str]:
    """Decode gameplay header bytes 0–28 (confidence per POG422_DECODE + bench)."""
    if len(preamble) != BINGO_WAGER_PREAMBLE_LEN:
        raise ValueError(f"preamble len {len(preamble)} != {BINGO_WAGER_PREAMBLE_LEN}")
    tid = preamble[3] | (preamble[4] << 8)
    game = preamble[5] | (preamble[6] << 8)
    ext_u32 = int.from_bytes(preamble[17:21], "little") if len(preamble) >= 21 else 0
    return {
        "soh": preamble[0],
        "msg_type": preamble[1],
        "direction": preamble[2],
        "tid": tid,
        "tid_hex": f"{tid:04X}",
        "game_id": game,
        "game_id_hex": f"{game:04X}",
        "play_sequence": preamble[7],
        "param_byte8": preamble[8],
        "bytes_9_12": preamble[9:13].hex(),
        "ext_u32_le_17": ext_u32,
        "bytes_21_28": preamble[21:29].hex() if len(preamble) >= 29 else "",
    }


def bingo_wager_wire_len_for_play_count(play_cards: int) -> int:
    """Expected wire length for N PLAY cards on T340+ (80 B fixed frame)."""
    if 1 <= play_cards <= BINGO_CARD_COUNT:
        return BINGO_WAGER_WIRE_LEN
    raise ValueError(f"bingo play card count {play_cards} out of range 1–{BINGO_CARD_COUNT}")


def bingo_wager_byte8_play_credits(wager: bytes) -> int:
    """Byte 8 on T340+ 80 B wager — PLAY credits (4/8/12/16 for 1–4 cards @ 25¢)."""
    if len(wager) <= 8:
        return 0
    return wager[8]


def bingo_card_raw_blocks_from_wager(wager: bytes) -> tuple[bytes, ...]:
    """Raw 12-byte card slots from 0x30 wager @ byte 29 (no grid decode)."""
    wl = len(wager)
    if wl == BINGO_WAGER_CENSUS_WIRE_LEN:
        slot_count = 2
    elif wl >= BINGO_WAGER_CARD_OFFSET + BINGO_CARD_BYTE_LEN:
        slot_count = BINGO_CARD_COUNT
    else:
        return tuple()
    end = BINGO_WAGER_CARD_OFFSET + slot_count * BINGO_CARD_BYTE_LEN
    if wl < end:
        return tuple()
    return tuple(
        wager[off : off + BINGO_CARD_BYTE_LEN]
        for off in (
            BINGO_WAGER_CARD_OFFSET + card * BINGO_CARD_BYTE_LEN
            for card in range(slot_count)
        )
    )


def bingo_play_slots_on_multi_wager(wager: bytes) -> int:
    """Count non-OFF card slots on an 80 B wager."""
    blocks = bingo_card_raw_blocks_from_wager(wager)
    if not blocks:
        return 0
    return sum(1 for block in blocks if not bingo_card_is_off(block))


def bingo_play_card_count_from_wager(wager: bytes) -> int:
    """PLAY cards implied by wager (byte 8 on T340+ 80 B, else slot census)."""
    wl = len(wager)
    if wl == BINGO_WAGER_WIRE_LEN:
        credits = bingo_wager_byte8_play_credits(wager)
        if credits in (0x04, 0x08, 0x0C, 0x10):
            return credits // 4
        return bingo_play_slots_on_multi_wager(wager)
    if wl == BINGO_WAGER_CENSUS_WIRE_LEN:
        return 2
    return 0


def bingo_result_wire_lens() -> tuple[int, ...]:
    return wire_lengths_for_cmd(BINGO_RESULT_MSG)


def bingo_calls_from_result(result: bytes) -> tuple[int, ...]:
    """Ball-call sequence from 0x39 result (T340 74 B wire, offset 21)."""
    if len(result) < BINGO_RESULT_CALL_OFFSET + 3:
        return tuple()
    end = len(result) - 3
    calls: list[int] = []
    for idx in range(BINGO_RESULT_CALL_OFFSET, end):
        if len(calls) >= BINGO_RESULT_CALL_COUNT:
            break
        val = result[idx]
        if 1 <= val <= 75:
            calls.append(val)
        elif calls:
            break
    return tuple(calls)


# Operator-verified Set A card 1 (2026-07-04 seq 7 bench) — row-major grid[row][col].
BINGO_LAB_SETA_CARD1_GRID: BingoGrid = (
    (4, 22, 42, 51, 75),
    (7, 30, 36, 48, 74),
    (8, 28, None, 53, 64),
    (1, 18, 43, 55, 65),
    (15, 25, 44, 49, 72),
)
# LP meter on SuperGold Bingo bench (~$7488.50) — BE32 cents @ bytes 9–16.
BINGO_LAB_DEFAULT_JPOT_CENTS = 748_850


def bingo_numbers_from_grid(grid: BingoGrid) -> tuple[int, ...]:
    """Row-major card numbers 1–75, skipping FREE center."""
    nums: list[int] = []
    for row in range(5):
        for col in range(5):
            if row == BINGO_CENTER_ROW and col == BINGO_CENTER_COL:
                continue
            val = grid[row][col]
            if val is None:
                raise ValueError("bingo grid cell missing number")
            if not (1 <= val <= 75):
                raise ValueError(f"bingo number out of range: {val}")
            nums.append(val)
    return tuple(nums)


def bingo_coverall_calls_from_grid(
    grid: BingoGrid,
    *,
    pad_to: int = BINGO_RESULT_CALL_COUNT,
) -> tuple[int, ...]:
    """Coverall call order: all 24 card numbers first, then unused 1–75."""
    card_nums = bingo_numbers_from_grid(grid)
    used = set(card_nums)
    calls = list(card_nums)
    for n in range(1, 76):
        if len(calls) >= pad_to:
            break
        if n not in used:
            calls.append(n)
    if len(calls) < pad_to:
        return tuple()
    return tuple(calls[:pad_to])


def build_bingo_jpot_synth(
    template: bytes,
    *,
    calls: tuple[int, ...] | list[int] | None = None,
    jpot_amount: int | None = None,
) -> bytes | None:
    """Build lab Bingo x9 (0x39) with forced coverall calls + progressive jackpot.

    Default calls: Set A card 1 row-major (24 numbers) padded to 50 with unused balls.
    Default progwinval: ``BINGO_LAB_DEFAULT_JPOT_CENTS`` ($7488.50 LP bench meter).

    NEVER use in production — corrupts Master/slave progressive accounting if abused.
    """
    if len(template) not in bingo_result_wire_lens():
        return None
    if len(template) < BINGO_RESULT_CALL_OFFSET + BINGO_RESULT_CALL_COUNT:
        return None
    draw = tuple(calls) if calls is not None else bingo_coverall_calls_from_grid(
        BINGO_LAB_SETA_CARD1_GRID
    )
    if len(draw) != BINGO_RESULT_CALL_COUNT:
        return None
    for ball in draw:
        if not (1 <= ball <= 75):
            return None
    frame = bytearray(template)
    frame[PROGRESSIVE_ACTIVEWINS_OFFSET] = 1
    amount = (
        int(jpot_amount)
        if jpot_amount is not None
        else BINGO_LAB_DEFAULT_JPOT_CENTS
    )
    be4 = amount.to_bytes(4, "big")
    frame[PROGRESSIVE_PROGDSP_OFFSET : PROGRESSIVE_PROGDSP_OFFSET + 4] = be4
    frame[PROGRESSIVE_PROGWIN_OFFSET : PROGRESSIVE_PROGWIN_OFFSET + 4] = be4
    for i, ball in enumerate(draw):
        frame[BINGO_RESULT_CALL_OFFSET + i] = ball & 0xFF
    pog422_set_frame_checksum(frame)
    return bytes(frame)


# Triple 7's 8-liner (POG bus id 970C, wager 0x80 / result 0x89 / ack 0x81).
TRIPLE7S_GAME_ID = "970C"
SLOT_WAGER_MSG = 0x80
SLOT_RESULT_MSG = 0x89
SLOT_ACK_MSG = 0x81
SLOT_89_WIRE_LEN = 36
SLOT_89_GRID_OFFSET = 21
SLOT_89_GRID_LEN = 9
SLOT_WIRE_TRIPLE7 = 0x0D
# Bench 2026-07-04 (operator oracle): 0x80 byte 8 = total credits when all 8 lines active.
#   0x08 (8)  = 1 credit × 8 lines — minimum for progressive jackpot tier
#   0x10 (16) = 2 credits × 8 lines — MAX bet ($4.00 @ $0.25/credit)
SLOT_WAGER_PROG_MIN_CREDITS = 0x08
SLOT_WAGER_MAX_CREDITS = 0x10

# Payline 1 = middle row (indices 3, 4, 5). Fillers avoid accidental 3-of-a-kind wins.
TRIPLE7S_LAB_JPOT_GRID: tuple[int, ...] = (
    0x00,
    0x04,
    0x05,  # row0 CHERRY LEMON PLUM
    0x0D,
    0x0D,
    0x0D,  # row1 TRIPLE7 ×3 (payline 1)
    0x06,
    0x09,
    0x08,  # row2 ORANGE BELL APPLE
)


def slot_result_wire_lens() -> tuple[int, ...]:
    return wire_lengths_for_cmd(SLOT_RESULT_MSG)


def slot_wager_byte8_play_credits(wager: bytes) -> int | None:
    """Total play credits from 0x80 byte 8 (when all 8 lines are in play)."""
    if len(wager) < 9 or wager[1] != SLOT_WAGER_MSG:
        return None
    return wager[8] & 0xFF


def slot_wager_progressive_eligible(wager: bytes) -> bool:
    """True when wager meets progressive minimum (1 credit × 8 lines = 8 credits)."""
    credits = slot_wager_byte8_play_credits(wager)
    return credits is not None and credits >= SLOT_WAGER_PROG_MIN_CREDITS


def slot_wager_is_max_bet(wager: bytes) -> bool:
    """True when wager byte 8 is MAX bet (2 credits × 8 lines = 16 credits)."""
    credits = slot_wager_byte8_play_credits(wager)
    return credits == SLOT_WAGER_MAX_CREDITS


def slot_grid_wire_from_result(result: bytes) -> tuple[int, ...] | None:
    if len(result) != SLOT_89_WIRE_LEN or result[1] != SLOT_RESULT_MSG:
        return None
    end = SLOT_89_GRID_OFFSET + SLOT_89_GRID_LEN
    return tuple(result[SLOT_89_GRID_OFFSET:end])


def build_triple7s_jpot_synth(
    template: bytes,
    *,
    grid: tuple[int, ...] | list[int] | None = None,
) -> bytes | None:
    """Build lab Triple 7's x9 (0x89): 3× TRIPLE7 on payline 1 (middle row).

    No progressive block on slot x9 — jackpot is detected from the symbol grid only.
    NEVER use in production.
    """
    if len(template) not in slot_result_wire_lens():
        return None
    draw = tuple(grid) if grid is not None else TRIPLE7S_LAB_JPOT_GRID
    if len(draw) != SLOT_89_GRID_LEN:
        return None
    frame = bytearray(template)
    for i, sym in enumerate(draw):
        frame[SLOT_89_GRID_OFFSET + i] = sym & 0xFF
    pog422_set_frame_checksum(frame)
    return bytes(frame)


def bingo_spot_indices() -> tuple[int, ...]:
    """Row-major 0–24, skipping FREE center (spot 12)."""
    return tuple(i for i in range(25) if i != BINGO_CENTER_ROW * 5 + BINGO_CENTER_COL)


def bingo_number_to_nibble(number: int) -> int:
    """Column-local nibble 0–14 (BINGO.C savebingcard hypothesis)."""
    return (number - 1) % 15


def bingo_nibble_to_number(col: int, nibble: int) -> int:
    return col * 15 + nibble + 1


def bingo_grid_to_card_bytes_legacy_cgs(grid: BingoGrid) -> bytes:
    """Pre-5.52U BINGO.C savebingcard layout — not valid on T340+ 5.52U bench captures."""
    if len(grid) != 5 or any(len(row) != 5 for row in grid):
        raise ValueError("bingo grid must be 5×5")
    nibbles: list[int] = []
    for row in range(5):
        for col in range(5):
            if row == BINGO_CENTER_ROW and col == BINGO_CENTER_COL:
                continue
            val = grid[row][col]
            if val is None:
                raise ValueError("FREE center must be None")
            nibbles.append(bingo_number_to_nibble(val))
    if len(nibbles) != 24:
        raise ValueError(f"expected 24 nibbles, got {len(nibbles)}")
    out = bytearray(BINGO_CARD_BYTE_LEN)
    for j, nib in enumerate(nibbles):
        out[j // 2] |= (nib & 0x0F) << ((j & 1) * 4)
    return bytes(out)


def bingo_card_bytes_to_grid_legacy_cgs(card_bytes: bytes) -> BingoGrid:
    """Pre-5.52U decode hypothesis — wrong grids on 5.52U/T340+ (use raw blocks until solved)."""
    if len(card_bytes) != BINGO_CARD_BYTE_LEN:
        raise ValueError(f"bingo card bytes len {len(card_bytes)} != {BINGO_CARD_BYTE_LEN}")
    if bingo_card_is_off(card_bytes):
        return tuple(tuple(None for _ in range(5)) for _ in range(5))
    nibbles: list[int] = []
    for j in range(24):
        nibbles.append((card_bytes[j // 2] >> ((j & 1) * 4)) & 0x0F)
    grid: list[list[int | None]] = [[0] * 5 for _ in range(5)]
    grid[BINGO_CENTER_ROW][BINGO_CENTER_COL] = None
    for ni, spot in enumerate(bingo_spot_indices()):
        row, col = divmod(spot, 5)
        grid[row][col] = bingo_nibble_to_number(col, nibbles[ni])
    return tuple(tuple(row) for row in grid)


def bingo_grid_to_card_bytes(grid: BingoGrid) -> bytes:
    """Alias — legacy CGS hypothesis until T340 codec is solved."""
    return bingo_grid_to_card_bytes_legacy_cgs(grid)


def bingo_card_bytes_to_grid(card_bytes: bytes) -> BingoGrid:
    """Alias — legacy CGS hypothesis until T340 codec is solved."""
    return bingo_card_bytes_to_grid_legacy_cgs(card_bytes)


def bingo_cards_from_wager(wager: bytes) -> tuple[BingoGrid, ...]:
    """Card grids via legacy hypothesis (see BINGO_CARD_CODEC_STATUS)."""
    blocks = bingo_card_raw_blocks_from_wager(wager)
    return tuple(bingo_card_bytes_to_grid_legacy_cgs(block) for block in blocks)


def bingo_card_is_off(card_bytes: bytes) -> bool:
    if len(card_bytes) != BINGO_CARD_BYTE_LEN:
        return True
    if all(b == 0 for b in card_bytes):
        return True
    if all(b == BINGO_CARD_OFF_LEGACY for b in card_bytes):
        return True
    return False


def bingo_grid_number_set(grid: BingoGrid) -> frozenset[int]:
    return frozenset(
        val
        for row in grid
        for val in row
        if val is not None
    )


def bingo_operator_reference_overlap_score(
    wager: bytes,
    reference: tuple[BingoGrid, ...] = BINGO_OPERATOR_REFERENCE_CARDS,
) -> int:
    """Count matching absolute numbers (best slot→ref assignment, legacy decode)."""
    blocks = bingo_card_raw_blocks_from_wager(wager)
    if len(blocks) != len(reference):
        return 0
    sets = [
        bingo_grid_number_set(bingo_card_bytes_to_grid_legacy_cgs(block))
        for block in blocks
        if not bingo_card_is_off(block)
    ]
    if len(sets) != len(reference):
        return 0
    ref_sets = [bingo_grid_number_set(grid) for grid in reference]
    best = 0
    for perm in itertools.permutations(range(len(sets))):
        total = sum(len(sets[perm[i]] & ref_sets[i]) for i in range(len(ref_sets)))
        if total > best:
            best = total
    return best


def bingo_grids_equal(a: BingoGrid, b: BingoGrid) -> bool:
    if len(a) != len(b):
        return False
    for row_a, row_b in zip(a, b):
        if len(row_a) != len(row_b):
            return False
        for va, vb in zip(row_a, row_b):
            if va != vb:
                return False
    return True


def bingo_grids_match_reference(
    decoded: tuple[BingoGrid, ...],
    reference: tuple[BingoGrid, ...],
) -> bool:
    if len(decoded) != len(reference):
        return False
    return all(bingo_grids_equal(d, r) for d, r in zip(decoded, reference))


# Lab-only Draw Poker royal flush synth (Jacks or Better @ 9400, msg 0x59 / 0x5A).
POKER_RESULT_MSG = 0x59
POKER_GRANT_MSG = 0x5A
SKILL_JPOT_GRANT_MSG_TYPES = frozenset({0x5A, 0x7A})
DRAW_POKER_GAME_ID = "9400"
POKER_CARD_DEAL_START = 21
POKER_CARD_DRAW_START = 26
POKER_CARD_SLOT_COUNT = 5
# A♥ K♥ Q♥ J♥ 10♥ — wire bytes from poker_card_wire_codec (bench 2026-07-02).
POKER_LAB_ROYAL_FLUSH_HEARTS_WIRE = (0x01, 0x0D, 0x0C, 0x0B, 0x0A)
# Lab filler draw-pile bytes when Master's 26–30 collide with forced deal (2♣…9♣).
POKER_LAB_DRAW_FILLER_WIRE = (0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19)


def poker_card_wires_in_frame(frame: bytes) -> tuple[int, ...]:
    return tuple(frame[POKER_CARD_DEAL_START : POKER_CARD_DRAW_START + POKER_CARD_SLOT_COUNT])


def poker_card_wires_unique(wires: tuple[int, ...]) -> bool:
    return len(wires) == len(set(wires))


def poker_synth_draw_bytes(
    template: bytes,
    deal_wires: tuple[int, ...],
) -> tuple[int, ...] | None:
    """Build draw-pile bytes 26–30: prefer Master order, de-dupe against deal."""
    if len(deal_wires) != POKER_CARD_SLOT_COUNT:
        return None
    used = set(deal_wires)
    out: list[int] = []
    fillers = iter(POKER_LAB_DRAW_FILLER_WIRE)
    start = POKER_CARD_DRAW_START
    end = start + POKER_CARD_SLOT_COUNT
    for b in template[start:end]:
        if b not in used:
            out.append(b & 0xFF)
            used.add(b & 0xFF)
            continue
        while True:
            try:
                filler = next(fillers)
            except StopIteration:
                return None
            if filler not in used:
                out.append(filler)
                used.add(filler)
                break
    if len(out) != POKER_CARD_SLOT_COUNT:
        return None
    return tuple(out)


def build_poker_royal_flush_synth(
    template: bytes,
    *,
    cards: tuple[int, ...] | list[int] | None = None,
) -> bytes | None:
    """Build lab Draw Poker 0x59: royal flush deal (21–25) + unique draw-pile (26–30).

    One 0x59 carries 10 distinct card wire bytes: initial deal plus the next five
    cards from the shoe (used when discards are replaced). Never duplicate deal
    cards into the draw-pile region.

    NEVER use in production.
    """
    if len(template) < 2 or template[1] != POKER_RESULT_MSG:
        return None
    if len(template) not in poker_result_wire_lens():
        return None
    end = POKER_CARD_DRAW_START + POKER_CARD_SLOT_COUNT
    if len(template) < end + 1:
        return None
    royal = tuple(cards) if cards is not None else POKER_LAB_ROYAL_FLUSH_HEARTS_WIRE
    if len(royal) != POKER_CARD_SLOT_COUNT:
        return None
    draw = poker_synth_draw_bytes(template, royal)
    if draw is None:
        return None
    frame = bytearray(template)
    for i, wire_byte in enumerate(royal):
        frame[POKER_CARD_DEAL_START + i] = wire_byte & 0xFF
    for i, wire_byte in enumerate(draw):
        frame[POKER_CARD_DRAW_START + i] = wire_byte & 0xFF
    if not poker_card_wires_unique(poker_card_wires_in_frame(frame)):
        return None
    pog422_set_frame_checksum(frame)
    return bytes(frame)


def poker_result_wire_lens() -> tuple[int, ...]:
    return wire_lengths_for_cmd(POKER_RESULT_MSG)


def poker_grant_wire_lens() -> tuple[int, ...]:
    return wire_lengths_for_cmd(POKER_GRANT_MSG)


# Lab-only SpinJack Flushjack synth (960F blackjack, msg 0x79).
BLACKJACK_RESULT_MSG = 0x79
SPINJACK_GAME_ID = "960F"
BJ_CARD_TABLE_START = 21
BJ_CARD_TABLE_COUNT = 4
BJ_CARD_SHOE_START = 25
BJ_CARD_SHOE_END = 39
# Player bytes 22 and 24 — suited A♥ + K♥ (flushjack in hearts).
SPINJACK_LAB_FLUSHJACK_HEARTS_WIRE = (0x01, 0x0D)
SPINJACK_LAB_SHOE_FILLER_WIRE = (
    0x12,
    0x13,
    0x14,
    0x15,
    0x16,
    0x17,
    0x18,
    0x19,
    0x1A,
    0x1B,
    0x1C,
    0x1D,
    0x22,
    0x23,
    0x24,
    0x25,
    0x26,
    0x27,
    0x28,
    0x29,
    0x2A,
    0x2B,
    0x2C,
    0x32,
    0x33,
    0x34,
    0x35,
    0x36,
    0x37,
    0x38,
    0x39,
    0x3A,
    0x3B,
    0x3C,
    0x3D,
    0x3E,
    0x3F,
)


def spinjack_card_wires_in_frame(frame: bytes) -> tuple[int, ...]:
    return tuple(frame[BJ_CARD_TABLE_START:BJ_CARD_SHOE_END])


def spinjack_card_wires_unique(wires: tuple[int, ...]) -> bool:
    return len(wires) == len(set(wires))


def _spinjack_next_filler(used: set[int], fillers) -> int | None:
    while True:
        try:
            filler = next(fillers)
        except StopIteration:
            return None
        if filler not in used:
            return filler
    return None


def spinjack_dealer_table_bytes(
    template: bytes,
    forced_player: tuple[int, int],
) -> tuple[int, int] | None:
    """Dealer hole (byte 21) and up (byte 23); keep Master unless collision."""
    used = set(forced_player)
    out: list[int] = []
    fillers = iter(SPINJACK_LAB_SHOE_FILLER_WIRE)
    for slot in (0, 2):
        b = template[BJ_CARD_TABLE_START + slot] & 0xFF
        if b not in used:
            out.append(b)
            used.add(b)
            continue
        filler = _spinjack_next_filler(used, fillers)
        if filler is None:
            return None
        out.append(filler)
        used.add(filler)
    return tuple(out)


def spinjack_synth_shoe_bytes(
    template: bytes,
    table_wires: tuple[int, ...],
) -> tuple[int, ...] | None:
    """Build shoe bytes 25–38: prefer Master order, de-dupe against table."""
    if len(table_wires) != BJ_CARD_TABLE_COUNT:
        return None
    used = set(table_wires)
    out: list[int] = []
    fillers = iter(SPINJACK_LAB_SHOE_FILLER_WIRE)
    for b in template[BJ_CARD_SHOE_START:BJ_CARD_SHOE_END]:
        if b not in used:
            out.append(b & 0xFF)
            used.add(b & 0xFF)
            continue
        filler = _spinjack_next_filler(used, fillers)
        if filler is None:
            return None
        out.append(filler)
        used.add(filler)
    expected = BJ_CARD_SHOE_END - BJ_CARD_SHOE_START
    if len(out) != expected:
        return None
    return tuple(out)


def build_spinjack_flushjack_synth(
    template: bytes,
    *,
    player_ace: int | None = None,
    player_ten: int | None = None,
) -> bytes | None:
    """Build lab SpinJack 0x79: suited ace + ten-value on player bytes 22 and 24.

    Table layout (bytes 21–24): D↓, P↑, D↑, P↑. Shoe 25–38 must be unique vs table.
    NEVER use in production.
    """
    if len(template) < 2 or template[1] != BLACKJACK_RESULT_MSG:
        return None
    if len(template) not in blackjack_result_wire_lens():
        return None
    if len(template) < BJ_CARD_SHOE_END + 1:
        return None
    ace = player_ace if player_ace is not None else SPINJACK_LAB_FLUSHJACK_HEARTS_WIRE[0]
    ten = player_ten if player_ten is not None else SPINJACK_LAB_FLUSHJACK_HEARTS_WIRE[1]
    forced_player = (ace & 0xFF, ten & 0xFF)
    if forced_player[0] == forced_player[1]:
        return None
    dealer = spinjack_dealer_table_bytes(template, forced_player)
    if dealer is None:
        return None
    table = (dealer[0], forced_player[0], dealer[1], forced_player[1])
    shoe = spinjack_synth_shoe_bytes(template, table)
    if shoe is None:
        return None
    frame = bytearray(template)
    for i, wire_byte in enumerate(table):
        frame[BJ_CARD_TABLE_START + i] = wire_byte & 0xFF
    for i, wire_byte in enumerate(shoe):
        frame[BJ_CARD_SHOE_START + i] = wire_byte & 0xFF
    if not spinjack_card_wires_unique(spinjack_card_wires_in_frame(frame)):
        return None
    pog422_set_frame_checksum(frame)
    return bytes(frame)


def blackjack_result_wire_lens() -> tuple[int, ...]:
    return wire_lengths_for_cmd(BLACKJACK_RESULT_MSG)


def build_poker_jpot_grant_synth(
    template: bytes,
    *,
    jpot_amount: int | None = None,
) -> bytes | None:
    """Build lab Draw Poker 0x5A skill progressive grant with forced progwinval.

    Skips when template already carries non-zero progwin BE (bytes 13–16).
    NEVER use in production.
    """
    if len(template) < 2 or template[1] != POKER_GRANT_MSG:
        return None
    if len(template) < PROGRESSIVE_PROGWIN_OFFSET + 4:
        return None
    if progressive_progwin_cents_be(template):
        return None
    frame = bytearray(template)
    frame[PROGRESSIVE_ACTIVEWINS_OFFSET] = 1
    amount = resolve_lab_progwin_cents(template, jpot_amount)
    be4 = amount.to_bytes(4, "big")
    frame[PROGRESSIVE_PROGDSP_OFFSET : PROGRESSIVE_PROGDSP_OFFSET + 4] = be4
    frame[PROGRESSIVE_PROGWIN_OFFSET : PROGRESSIVE_PROGWIN_OFFSET + 4] = be4
    pog422_set_frame_checksum(frame)
    return bytes(frame)


def is_gameplay_ack(msg_type: int | None) -> bool:
    return msg_type is not None and msg_type in GAMEPLAY_ACK_MSG_TYPES


def is_skill_jpot_grant(msg_type: int | None) -> bool:
    """Master → slave skill progressive grant (Draw Poker 0x5A, slots 0x7A, …)."""
    return msg_type is not None and msg_type in SKILL_JPOT_GRANT_MSG_TYPES


def is_gameplay_request(msg_type: int | None) -> bool:
    return msg_type is not None and msg_type in GAMEPLAY_REQUEST_MSG_TYPES


def uart_window_lead_is_response_echo(first_byte: int | None) -> bool:
    """True when SLV-IN burst starts on x9 opcode (inject echo / mis-aligned glue)."""
    return first_byte is not None and first_byte in GAMEPLAY_RESPONSE_MSG_TYPES


def uart_window_lead_is_native_wager(first_byte: int | None) -> bool:
    """True when SLV-IN burst starts with a gameplay request (native wager HS)."""
    return first_byte is not None and (
        first_byte == SOH or first_byte in GAMEPLAY_REQUEST_MSG_TYPES
    )


def uart_echo_false_glued_wager(
    msg_type: int | None,
    *,
    packets_in_window: int,
    uart_first_byte: int | None,
) -> bool:
    """Glued-window false pc70 from x9-inject echo tail — not a real native wager."""
    return (
        packets_in_window >= 2
        and uart_window_lead_is_response_echo(uart_first_byte)
        and is_gameplay_request(msg_type)
    )


def is_gameplay_response(msg_type: int | None) -> bool:
    return msg_type is not None and msg_type in GAMEPLAY_RESPONSE_MSG_TYPES


def is_progressive_broadcast(msg_type: int | None) -> bool:
    """Master progressive entity poll (tid=0000 on wire) — safe to defer mid-round."""
    return msg_type == 0x0A


def is_bank_progressive_jackpot_frame(frame: bytes | None) -> bool:
    """Winner-TID frame carrying progwinval > 0 — bank progressive jackpot issue.

    Two Master-origin shapes qualify:
      * gameplay outcome x9 (0x29/0x39/0x49/0x69, and 0x59/0x79 when stripped)
      * skill jpot grant 0x5A/0x7A answering a slave claim (0x52/0x72) — this is
        where Draw Poker royals and SpinJack bonus progressives carry the amount

    Every seat eventually needs the frame for bank bells, but under EDGE the
    bridge **defers multiport eth fanout until Master has the winner's x1**
    (EG 3.0.1.31+): first hop is winner-only unicast so non-winner bus traffic
    does not race the progressive close. Full bank flood is a later step.

    The slave claim itself (0x52/0x72) is not included: it carries no progwinval
    and is addressed to the Master, so it needs no bank fanout.
    """
    if not frame:
        return False
    meta = parse_packet_meta(frame)
    if not (
        is_gameplay_response(meta.msg_type) or is_skill_jpot_grant(meta.msg_type)
    ):
        return False
    return progressive_progwin_cents_be(frame) > 0


# eth→UART inject priority (higher = sooner). Same class stays FIFO.
INJECT_PRIORITY_NORMAL = 0
INJECT_PRIORITY_STOM = 10
# SCR 0x03/0x0B above soft POG_UPDT 0x0A / other STOM (EG 4.1.0.2).
INJECT_PRIORITY_SCR = 12
# Slave x1 after Master bank progressive (must beat lab synth / foreign 80).
INJECT_PRIORITY_BANK_JPOT_X1 = 20
# Local status synth 0D close path.
INJECT_PRIORITY_P0_CLOSE = 30

# Hub/spoke: drop duplicate SCR eth→UART while same TID is in-flight or freshly answered.
SCR_COALESCE_MS = 220.0

# How long after Master bank jpot issue we treat that TID's x1 as head-of-queue.
BANK_JPOT_X1_PRIORITY_HOLD_MS = 30_000
# Max eth→UART re-injects of the winner x1 while armed and Master has not copper-PC'd it.
BANK_JPOT_X1_MAX_REINJECT = 12
# After copper-proven x1, keep last ack only for Master progressive re-TX race.
# T340 progressive close aborts in ~200 ms — 10 s was useless; sealed reinject
# must land inside that abort window (soak: 2nd 49 @ ~163 ms, 3rd @ ~230 ms).
BANK_JPOT_X1_SEAL_MS = 250
# After x1 closes without a recent result x9 for that TID, refuse progressive
# award inject on the local terminal (empty hand → Jackpot Won race).
BANK_JPOT_EMPTY_CLOSE_BLOCK_MS = 15_000
# Ignore result x9 older than this when classifying "x1 without result".
BANK_JPOT_RESULT_X9_STALE_MS = 2_000


def bank_progressive_jackpot_winner_tid(frame: bytes | None) -> str | None:
    """Uppercase winner TID when frame is a bank progressive jackpot issue, else None."""
    if not is_bank_progressive_jackpot_frame(frame):
        return None
    meta = parse_packet_meta(frame)
    tid = (meta.tid_hex or "").strip().upper()
    return tid or None


def is_bank_jpot_x1_priority_ack(
    msg_type: int | None,
    tid_hex: str | None,
    armed_until_ns: dict[str, int] | None,
    *,
    now_ns: int | None = None,
) -> bool:
    """True when this game ACK is for a TID armed after Master bank progressive jpot."""
    if not is_gameplay_ack(msg_type):
        return False
    if not tid_hex or not armed_until_ns:
        return False
    tid_u = tid_hex.strip().upper()
    exp = armed_until_ns.get(tid_u)
    if exp is None:
        return False
    if now_ns is None:
        now_ns = time.perf_counter_ns()
    return now_ns <= exp


def is_gameplay_close_leg(
    msg_type: int | None,
    *,
    source: str | None = None,
) -> bool:
    """Gameplay transaction close legs: any game ACK (x1) or status 0x0D.

    Bank progressive exclusive arm may reorder / prioritize winner x1, but must
    **never** transaction-defer these for any TID (EG 3.0.1.33+). Deferring a
    non-winner slave's x1 or synth/native 0x0D causes MRT and post-jpot CGS.
    """
    if source == "status_0d_synth":
        return True
    if is_gameplay_ack(msg_type):
        return True
    return msg_type == 0x0D


PROGRESSIVE_0A_GAME_COUNT = 16
PROGRESSIVE_0A_PENNIES_OFFSET = 15
PROGRESSIVE_CSV_NAME = "progressives.csv"
DEFAULT_ETHERGOLD_ROOT = "/home/topbox/EtherGold"


def resolve_progressives_csv_path(explicit: str | None = None) -> str:
    """Path for DyzineFrame-compatible progressives.csv (atomic writes, mtime reload)."""
    if explicit:
        return explicit
    env_path = os.environ.get("ETHERGOLD_PROGRESSIVES_CSV")
    if env_path:
        return env_path
    root = os.environ.get("ETHERGOLD_ROOT", DEFAULT_ETHERGOLD_ROOT)
    return os.path.join(root, PROGRESSIVE_CSV_NAME)


def decode_progressive_0a(frame: bytes) -> dict[int, int] | None:
    """Decode Master 0x0A broadcast into game_id (1..16) → pennies (uint32 LE at byte 15).

    Wire byte 2 is Bank Number (same as every SOH opcode); callers must filter
    foreign banks via ``local_bank_accepts`` / ``ProgressiveCsvWriter.maybe_update``.
    """
    wire_len = COMMAND_WIRE_LEN.get(0x0A, 82)
    if len(frame) != wire_len:
        return None
    if frame[0] != SOH or frame[1] != 0x0A or frame[-1] != EOP:
        return None
    if not pog422_frame_checksum_ok(frame):
        return None
    if tid_le(frame, 3) != BANK_PROGRESSIVE_BROADCAST_TID:
        return None
    off = PROGRESSIVE_0A_PENNIES_OFFSET
    end_before_crc = len(frame) - 3
    pennies_by_game: dict[int, int] = {}
    for slot in range(PROGRESSIVE_0A_GAME_COUNT):
        if off + 4 > end_before_crc:
            return None
        game_id = slot + 1
        pennies = int.from_bytes(frame[off : off + 4], "little")
        off += 4
        if game_id == 15:
            pennies = 0
        pennies_by_game[game_id] = pennies
    return pennies_by_game


def progressive_0a_dedupe_key(pennies_by_game: dict[int, int]) -> tuple[int, ...]:
    return tuple(pennies_by_game.get(game_id, 0) for game_id in range(1, 17))


def write_progressives_csv(path: str, pennies_by_game: dict[int, int]) -> None:
    """Write full 1..16 rows atomically (temp + replace) for OHD mtime-based reload."""
    parent = os.path.dirname(path)
    if parent:
        os.makedirs(parent, exist_ok=True)
    stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    fd, tmp_path = tempfile.mkstemp(
        dir=parent or ".",
        prefix=".progressives.",
        suffix=".tmp",
    )
    try:
        with os.fdopen(fd, "w", newline="", encoding="utf-8") as handle:
            writer = csv.writer(handle)
            writer.writerow(["game_id", "pennies", "updated_utc"])
            for game_id in range(1, 17):
                pennies = pennies_by_game.get(game_id, 0)
                if game_id == 15:
                    pennies = 0
                writer.writerow([game_id, pennies, stamp])
        os.replace(tmp_path, path)
        try:
            os.chmod(path, 0o644)
            real_user = os.environ.get("ETHERGOLD_REAL_USER", "topbox")
            import pwd

            pw = pwd.getpwnam(real_user)
            os.chown(path, pw.pw_uid, pw.pw_gid)
        except (OSError, KeyError, ImportError):
            try:
                os.chmod(path, 0o644)
            except OSError:
                pass
    except Exception:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise


# Heartbeat log of on-disk meters even when 0A pennies are unchanged
# (or only non-logged games tick) — enables non-reset detection in EVAL.
PROGRESSIVES_CSV_SNAPSHOT_S = 60.0


def progressive_pennies_event_fields(pennies_by_game: dict[int, int]) -> dict[str, int]:
    """game1..game16 pennies for event_log (always 16 keys)."""
    out: dict[str, int] = {}
    for game_id in range(1, 17):
        pennies = int(pennies_by_game.get(game_id, 0) or 0)
        if game_id == 15:
            pennies = 0
        out[f"game{game_id}"] = pennies
    return out


def read_progressives_csv(path: str) -> dict[int, int] | None:
    """Load game_id → pennies from a DyzineFrame-compatible progressives.csv."""
    try:
        with open(path, encoding="utf-8", newline="") as handle:
            reader = csv.DictReader(handle)
            rows: dict[int, int] = {}
            for row in reader:
                try:
                    gid = int(row.get("game_id") or 0)
                    pennies = int(row.get("pennies") or 0)
                except (TypeError, ValueError):
                    continue
                if 1 <= gid <= 16:
                    rows[gid] = 0 if gid == 15 else pennies
            if not rows:
                return None
            for game_id in range(1, 17):
                rows.setdefault(game_id, 0)
            rows[15] = 0
            return rows
    except OSError:
        return None


class ProgressiveCsvWriter:
    """Master 0x0A → progressives.csv with content dedupe (ETH + UART relay duplicates)."""

    def __init__(
        self,
        csv_path: str | None = None,
        *,
        event_log: EventLog | None = None,
        snapshot_interval_s: float = PROGRESSIVES_CSV_SNAPSHOT_S,
    ):
        self.csv_path = resolve_progressives_csv_path(csv_path)
        self._event_log = event_log
        self._lock = threading.Lock()
        self._last_key: tuple[int, ...] | None = None
        self._last_decoded: dict[int, int] | None = None
        self._last_source: str | None = None
        self._snapshot_interval_s = max(1.0, float(snapshot_interval_s))
        self._last_snapshot_mono: float = 0.0
        self._logged_0a_reject = False

    def maybe_update(self, frame: bytes, *, source: str) -> bool:
        frame_bank = wire_bank_number(frame)
        if not local_bank_accepts(frame_bank):
            if self._event_log is not None:
                self._event_log.emit(
                    "progressive_0a_bank_skip",
                    source=source,
                    path=self.csv_path,
                    pkt_len=len(frame),
                    tid=tid_le(frame, 3),
                    **packet_bank_event_fields(frame_bank),
                )
            return False
        decoded = decode_progressive_0a(frame)
        if decoded is None:
            if (
                self._event_log is not None
                and not self._logged_0a_reject
                and len(frame) >= 2
                and frame[1] == 0x0A
            ):
                self._logged_0a_reject = True
                ck = pog422_frame_checksum_ok(frame) if len(frame) >= 4 else False
                self._event_log.emit(
                    "error",
                    where="progressive_0a_decode",
                    message="decode_progressive_0a returned None",
                    source=source,
                    path=self.csv_path,
                    pkt_len=len(frame),
                    checksum_ok=ck,
                    tid=tid_le(frame, 3),
                    hex_full=bytes_hex_line(frame),
                )
                try:
                    Path("/tmp/eg_0a_reject.hex").write_text(
                        bytes_hex_line(frame) + "\n",
                        encoding="ascii",
                    )
                except OSError:
                    pass
            return False
        key = progressive_0a_dedupe_key(decoded)
        with self._lock:
            self._last_decoded = dict(decoded)
            self._last_source = source
            # DF / HTTP boot-clear, or a deleted CSV, must not stick forever
            # because 0x0A pennies often do not change for hours.
            if key == self._last_key and os.path.isfile(self.csv_path):
                try:
                    if os.path.getsize(self.csv_path) > 0:
                        return False
                except OSError:
                    pass
            self._last_key = key
        try:
            write_progressives_csv(self.csv_path, decoded)
        except OSError as exc:
            if self._event_log is not None:
                self._event_log.emit(
                    "error",
                    where="progressive_0a_csv",
                    message=str(exc),
                    path=self.csv_path,
                    source=source,
                )
            return False
        if self._event_log is not None:
            self._event_log.emit(
                "progressive_0a_csv_update",
                source=source,
                path=self.csv_path,
                **packet_bank_event_fields(frame_bank),
                **progressive_pennies_event_fields(decoded),
            )
        try:
            from machine_status_ipc import notify_progressives_changed

            notify_progressives_changed()
        except Exception:
            pass
        return True

    def snapshot_tick(self, *, now_mono: float | None = None) -> bool:
        """Emit progressives_csv_snapshot at most once per snapshot_interval_s.

        Uses last 0A decode when available; else reads the on-disk CSV. Always
        logs all game1..game16 pennies so EVAL can see freeze / no reseed.
        """
        if self._event_log is None:
            return False
        now = float(now_mono if now_mono is not None else time.monotonic())
        with self._lock:
            if (
                self._last_snapshot_mono > 0.0
                and (now - self._last_snapshot_mono) < self._snapshot_interval_s
            ):
                return False
            self._last_snapshot_mono = now
            decoded = dict(self._last_decoded) if self._last_decoded else None
            last_src = self._last_source
        if decoded is None:
            decoded = read_progressives_csv(self.csv_path)
            source = "csv_file"
        else:
            source = last_src or "last_0a"
        if decoded is None:
            self._event_log.emit(
                "progressives_csv_snapshot",
                path=self.csv_path,
                source="missing",
                ok=0,
            )
            return False
        self._event_log.emit(
            "progressives_csv_snapshot",
            path=self.csv_path,
            source=source,
            ok=1,
            **progressive_pennies_event_fields(decoded),
        )
        return True


def status_0d_counter_forward(a: int, b: int) -> bool:
    """True when b is same or ahead of a on the 8-bit close counter ring."""
    return ((b - a) & 0xFF) < 128


def note_status_0d_global_counter(current: int | None, observed: int) -> int:
    """Track the highest forward close counter seen (bank-wide, 8-bit wrap)."""
    if current is None:
        return observed & 0xFF
    if status_0d_counter_forward(current, observed):
        return observed & 0xFF
    return current


ETH_INJECT_TID_BLACKLIST_MAX = 100


def eth_inject_tid_blacklist_add(
    blacklist: set[str],
    order: list[str],
    tid_hex: str | None,
    *,
    max_size: int = ETH_INJECT_TID_BLACKLIST_MAX,
) -> bool:
    """Record TID from an ETH-origin frame queued for UART inject.

    UART capture after inject is multi-drop echo/response — not local T340+ origin.
    """
    if not tid_hex:
        return False
    key = tid_hex.upper()
    if key in blacklist:
        return False
    while len(order) >= max_size:
        oldest = order.pop(0)
        blacklist.discard(oldest)
    blacklist.add(key)
    order.append(key)
    return True


def local_slave_tid_uart_learn_ok(
    tid_hex: str | None,
    msg_type: int | None = None,
    *,
    eth_inject_tid_blacklist: set[str] | None = None,
) -> bool:
    """Whether the first UART→ETH origin frame may seed session local_slave_tid.

    Typical first origins: Transfer Master Config (0x02/0x03), gameplay wager (x0),
    or ack (x1). Rejects bank progressive (0x0A tid=0000) and ETH-inject echo TIDs.
    """
    if not tid_hex:
        return False
    key = tid_hex.upper()
    if key == BANK_PROGRESSIVE_BROADCAST_TID:
        return False
    if msg_type == 0x0A:
        return False
    if eth_inject_tid_blacklist and key in eth_inject_tid_blacklist:
        return False
    return True


def status_synth_x1_trigger_ok(
    meta: PacketMeta,
    local_tid: str | None = None,
) -> tuple[bool, str | None]:
    """Whether slave x1 ACK in this HS window should arm local status-close synth.

    Copper: solo x1 window. Bridge: x9 inject echo often glues with x1 — accept x1
    when it is the last packet in the window (typical x9+x1 glue). Local terminal x1
    also arms when trailing bus packets (e.g. 0x0A poll) glue into the same window.
    """
    if not is_gameplay_ack(meta.msg_type):
        return False, None
    if not meta.game_id_hex:
        return False, None
    n = meta.packet_count_in_window
    idx = meta.index_in_window
    is_local = (
        local_tid is not None
        and meta.tid_hex
        and meta.tid_hex.upper() == local_tid.upper()
    )
    if n == 1:
        return True, "solo_x1"
    if n >= 2 and idx == n - 1:
        return True, "x1_last_in_window"
    if is_local:
        return True, "local_x1_glued"
    return False, None

DATA_DIR = "data"
LOGS_DIR = "logs"
PACKETS_RAW = "packets.raw"
EVENTS_JSONL = "events.jsonl"
# Bus TX echo after eth→UART inject (what RX FIFO saw while we drove the line).
INJECT_ECHO_JSONL = "inject_echo.jsonl"
HW_DEBUG_JSONL = "hw_debug.jsonl"
PROD_ERRORS_JSONL = "errors.jsonl"

# Production mode: errors.jsonl only — no packets.raw, no events.jsonl flood.
PROD_LOG_EVENTS = frozenset(
    {
        "session_start",
        "session_stop",
        "local_slave_tid_learned",
        "local_slave_tid_learn_skip",
        "native_uart_status_close",
        "status_0d_synth_skip",
        "inject_hs_timeout",
        "window_skip",
        "inject_fail",
        "error",
        "serial_open_fail",
        "eth_open_fail",
        "link_arp_fail",
        "link_recover",
        "link_recover_ok",
        "link_recover_fail",
        "link_recover_escalate",
        "link_carrier_down",
        "link_carrier_up",
        "supervisor_start",
        "supervisor_stop",
        "supervisor_restart",
        "supervisor_usb_recover",
        "supervisor_disabled",
        "progressive_0a_csv_update",
    }
)
# Dev-light: faults + MRT/synth close diagnostics (events.jsonl); no packets.raw flood.
DEV_LIGHT_LOG_EVENTS = PROD_LOG_EVENTS | frozenset(
    {
        "x9_no_local_x1",
        "mrt_risk_late_0d_after_x1",
        "mrt_risk_arm_skip",
        "mrt_risk_await_0d_abandon",
        "await_local_x1_timeout",
        "mrt_x1_inferred_no_x9_retry",
        "mrt_x1_after_x9",
        "mrt_x9_inject_diag",
        "mrt_peer_x1_in_echo",
        "x1_harvest_path",
        "post_x9_inject_grace_absorb",
        "post_x9_inject_grace_arm",
        "post_de_soh_hold",
        "post_de_soh_glued_window",
        "post_x9_handoff_capture_reject",
        "inject_peer_uart_eth",
        "inject_peer_early_eth",
        "local_round_superseded",
        "status_0d_synth_trigger",
        "status_0d_synth_immediate",
        "status_0d_synth_queued",
        "status_0d_synth_done",
        "status_0d_synth_skip",
        "status_0d_synth_deferred_tx_lock",
        "mrt_x9_inject_slow",
        "status_0d_relay_synth_skip",
        "inject_deferred",
        "transaction_defer",
        "transaction_defer_flush",
        "transaction_defer_abandon",
        "trunk_round_stale_abandon",
        # WiFi-edge / MRT soak: Master outcome TX + same-seq retry crumbs
        "peer_gameplay_eth_fwd",
        # Lab synthetic T340+ seat (UART→ETH wager forge)
        "lab_gameplay_synth_armed",
        "lab_gameplay_session_start",
        "lab_gameplay_session_end",
        "lab_gameplay_synth_wager",
        "lab_gameplay_synth_ack",
        "lab_gameplay_synth_skip",
        "early_win_phase1_hit",
        "early_win_slot_hit",
        "early_win_slot_fireworks_ipc",
        "early_win_slot_fireworks_err",
        "early_win_keno_hit",
        "early_win_keno_fireworks_ipc",
        "early_win_keno_fireworks_err",
        "debug_log_rotate",
    }
)
DEV_LIGHT_INJECT_DEFER_REASONS = frozenset(
    {
        "await_local_0d",
        "await_local_x1",
        "hs_busy",
        "post_x9_grace_ois",
        "post_x9_grace_machine_relay",
    }
)
DEV_LIGHT_TRANSACTION_DEFER_REASONS = frozenset(
    {
        "await_local_0d",
        "game_transaction_gate",
        "0a_latest_wins",
        "queue_cap",
    }
)
# status_0d_synth_skip: log actionable synth failures only (not await_passthrough on Master).
PROD_SYNTH_SKIP_LOG_REASONS = frozenset({"no_x9", "x9_synth_disabled"})
DEV_LIGHT_SYNTH_SKIP_LOG_REASONS = PROD_SYNTH_SKIP_LOG_REASONS | frozenset(
    {"await_passthrough", "superseded_seq", "master_uart_x9_native_0d"}
)
# Config transfer opcodes — log wire_len_crop skips in prod (errors.jsonl is sparse).
PROD_CONFIG_TRANSFER_MSGS = frozenset({"01", "02", "03", "04", "09", "0A", "0B", "0E"})
FIXTURE_STATS = "fixture_stats.json"


def resolve_dev_light(*, cli_flag: bool = False, prod_mode: bool = False) -> bool:
    """Fault-only events.jsonl tier (between full debug and --prod)."""
    if prod_mode:
        return False
    if cli_flag:
        return True
    mode = (os.environ.get("FIXTURE_LOG_MODE") or "").strip().lower()
    if mode in ("dev_light", "dev-light", "light"):
        return True
    if mode in ("full", "debug", "prod"):
        return False
    return _env_bool("FIXTURE_DEV_LIGHT", False)


@dataclass(frozen=True)
class BusChannel:
    key: str
    default_port: str
    legacy_silk: str
    gpio_name: str


BUS_CHANNELS = {
    "machine": BusChannel("machine", "/dev/ttyXRUSB1", "SLV-IN", "GPIOB1"),
    "ois": BusChannel("ois", "/dev/ttyXRUSB0", "MST-IN", "GPIOA1"),
}


@dataclass(frozen=True)
class FixtureIdentity:
    fixture_pos: int
    pi_label: str = ""
    role: str = "listen"  # listen | bridge


@dataclass
class PacketMeta:
    raw: bytes
    msg_type: Optional[int] = None
    bank_number: Optional[int] = None
    tid_hex: Optional[str] = None
    game_id_hex: Optional[str] = None
    play_sequence: Optional[int] = None
    session_counter: Optional[int] = None
    master_tid_hex: Optional[str] = None
    packet_count_in_window: int = 1
    index_in_window: int = 0

    @property
    def msg_type_hex(self) -> str:
        if self.msg_type is None:
            return "??"
        return f"{self.msg_type:02X}"

    @property
    def hex_line(self) -> str:
        return ",".join(f"{b:02X}" for b in self.raw)

    @property
    def expected_response(self) -> Optional[int]:
        if self.msg_type is None:
            return None
        low = self.msg_type & 0x0F
        if low == 0:
            return REQUEST_TO_RESPONSE.get(self.msg_type)
        return None

    @property
    def expected_ack(self) -> Optional[int]:
        if self.msg_type is None:
            return None
        high = self.msg_type & 0xF0
        low = self.msg_type & 0x0F
        if low == 9:
            return high | 0x01
        return None


def now_ns() -> dict[str, int]:
    return {
        "t_perf_ns": time.perf_counter_ns(),
        "t_mono_ns": time.monotonic_ns(),
    }


def utc_iso() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="microseconds")


def rotate_file_if_large(path: Path, max_bytes: int) -> None:
    """Prod long-run: rotate + prune via prod_log_manager (flat runtime dir)."""
    if max_bytes <= 0:
        return
    prod_dir = Path(__file__).resolve().parent
    if str(prod_dir) not in sys.path:
        sys.path.insert(0, str(prod_dir))
    try:
        from prod_log_manager import manage_prod_log

        manage_prod_log(path, max_bytes=max_bytes, max_age_s=PROD_LOG_MAX_AGE_S)
    except ImportError:
        if not path.is_file():
            return
        try:
            if path.stat().st_size < max_bytes:
                return
        except OSError:
            return
        rotated = Path(f"{path}.1")
        if rotated.is_file():
            rotated.unlink()
        path.rename(rotated)


def resolve_channel(key: str) -> BusChannel:
    if key == "both":
        raise SystemExit(
            "CHANNEL=both needs resolve_bridge_channels() "
            "(bridge dual or listen dual-uartd)"
        )
    try:
        return BUS_CHANNELS[key]
    except KeyError as exc:
        raise SystemExit(f"Unknown channel {key!r}; use: {', '.join(BUS_CHANNELS)}") from exc


def resolve_bridge_channels(key: str) -> list[BusChannel]:
    """Bridge modes: machine, ois, or both (A+B on one Pi)."""
    if key == "both":
        return [BUS_CHANNELS["machine"], BUS_CHANNELS["ois"]]
    return [resolve_channel(key)]


def bridge_uses_eth_channel_tag(key: str) -> bool:
    return key == "both"


def resolve_output_dir(path: str) -> Path:
    candidate = Path(path)
    if not candidate.is_absolute():
        candidate = Path.cwd() / candidate
    candidate.mkdir(parents=True, exist_ok=True)
    return candidate


def resolve_logs_dir(path: str) -> Path:
    candidate = Path(path)
    if not candidate.is_absolute():
        candidate = Path.cwd() / candidate
    candidate.mkdir(parents=True, exist_ok=True)
    return candidate


def resolve_serial_port(channel: BusChannel, explicit: Optional[str]) -> str:
    """Resolve ttyXRUSB* for *channel*.

    Explicit path wins (CLI override). Otherwise:
    - ois → prefer ttyXRUSB0, else lowest numbered XRUSB node
    - machine → prefer ttyXRUSB1, else highest numbered XRUSB node
    (covers post-reenumerate 0+2 when XRUSB1 is gone).
    """
    if explicit:
        return explicit
    matches = sorted(glob.glob("/dev/ttyXRUSB*"))
    if not matches:
        if os.path.exists(channel.default_port):
            return channel.default_port
        raise SystemExit("Serial port /dev/ttyXRUSB* could not be located")
    if channel.key == "ois":
        preferred = "/dev/ttyXRUSB0"
        if preferred in matches:
            return preferred
        return matches[0]
    # machine (and any non-ois channel)
    preferred = "/dev/ttyXRUSB1"
    if preferred in matches:
        return preferred
    return matches[-1]


_UART_CONTENTION_UNITS = (
    "ethergold-collector.service",
    "ethergold.service",
    "fixture-collector.service",
    "fixture-prod.service",
)


def _cmdline_of(pid: int) -> str:
    try:
        raw = Path(f"/proc/{pid}/cmdline").read_bytes()
    except OSError:
        return ""
    return raw.replace(b"\0", b" ").decode("utf-8", errors="replace").strip()


def _holders_via_proc(port: str, *, ignore_pids: set[int]) -> list[dict[str, Any]]:
    """Scan /proc/*/fd for openers of *port* (works without fuser)."""
    holders: list[dict[str, Any]] = []
    try:
        want = os.path.realpath(port)
    except OSError:
        want = port
    for fd_path in glob.glob("/proc/[0-9]*/fd/[0-9]*"):
        try:
            target = os.readlink(fd_path)
        except OSError:
            continue
        if target != port and target != want:
            continue
        try:
            pid = int(fd_path.split("/")[2])
        except (IndexError, ValueError):
            continue
        if pid in ignore_pids:
            continue
        holders.append(
            {
                "pid": pid,
                "cmdline": _cmdline_of(pid),
                "via": "proc_fd",
                "port": port,
            }
        )
    return holders


def _holders_via_fuser(port: str, *, ignore_pids: set[int]) -> list[dict[str, Any]]:
    if not shutil.which("fuser"):
        return []
    try:
        proc = subprocess.run(
            ["fuser", "-v", port],
            check=False,
            capture_output=True,
            text=True,
            timeout=3,
        )
    except (OSError, subprocess.TimeoutExpired):
        return []
    # fuser prints PIDs on stdout; -v details on stderr.
    holders: list[dict[str, Any]] = []
    seen: set[int] = set()
    blob = f"{proc.stdout or ''}\n{proc.stderr or ''}"
    for token in blob.replace("\n", " ").split():
        if not token.isdigit():
            continue
        pid = int(token)
        if pid in ignore_pids or pid in seen:
            continue
        seen.add(pid)
        holders.append(
            {
                "pid": pid,
                "cmdline": _cmdline_of(pid),
                "via": "fuser",
                "port": port,
            }
        )
    return holders


def _unit_active(unit: str) -> bool:
    if not shutil.which("systemctl"):
        return False
    try:
        proc = subprocess.run(
            ["systemctl", "is-active", "--quiet", unit],
            check=False,
            timeout=3,
        )
    except (OSError, subprocess.TimeoutExpired):
        return False
    return proc.returncode == 0


def _is_pog422_uartd_holder(holder: dict[str, Any]) -> bool:
    """True when the port holder is our HS/RX daemon (socket client attaches to it)."""
    cmd = str(holder.get("cmdline") or "")
    base = os.path.basename(cmd.split()[0]) if cmd.split() else ""
    return "pog422_uartd" in cmd or base == "pog422_uartd"


def probe_uart_port_contention(
    ports: Sequence[str] | None = None,
    *,
    ignore_pids: Sequence[int] | None = None,
    prod_mode: bool = False,
    own_uartd: bool = False,
) -> dict[str, Any]:
    """Who owns ttyXRUSB* + which systemd units would steal listen RX.

    Tip/workshop listen must not share machine UART with ethergold-collector
    (prod). Dual-open can succeed and still leave tip with window_bytes=0.

    When ``own_uartd`` is True (listen will attach via pog422_uartd socket),
    holders that *are* pog422_uartd are expected — not foreign contention.
    Without that, uartd-persist / hat_listen start-uartd-then-python loops
    forever on ``foreign_pid_holds_port`` (Mini OHD .33 2026-09-15).
    """
    ignore = set(ignore_pids or ())
    ignore.add(os.getpid())
    if ports is None:
        ports = sorted(glob.glob("/dev/ttyXRUSB*"))
    port_list = [p for p in ports if p]
    holders: list[dict[str, Any]] = []
    for port in port_list:
        found = _holders_via_fuser(port, ignore_pids=ignore)
        if not found:
            found = _holders_via_proc(port, ignore_pids=ignore)
        holders.extend(found)

    if own_uartd:
        holders = [h for h in holders if not _is_pog422_uartd_holder(h)]

    units_active = [u for u in _UART_CONTENTION_UNITS if _unit_active(u)]
    # Field collector owning the port is expected when this process *is* prod.
    collector_active = any(
        u.startswith("ethergold-collector") or u.startswith("fixture-collector")
        for u in units_active
    )
    bridge_active = any(
        u in ("ethergold.service", "fixture-prod.service") for u in units_active
    )
    foreign_holders = bool(holders)
    # Prod listen IS the collector — do not treat its own unit as contention.
    unit_steal = (collector_active and not prod_mode) or bridge_active
    contention = foreign_holders or unit_steal
    reasons: list[str] = []
    if foreign_holders:
        reasons.append("foreign_pid_holds_port")
    if collector_active and not prod_mode:
        reasons.append("ethergold_collector_active")
    if bridge_active:
        reasons.append("ethergold_bridge_active")

    return {
        "uart_contention": bool(contention),
        "uart_contention_ok": not contention,
        "uart_contention_reasons": reasons,
        "uart_port_holders": holders,
        "uart_ports_checked": list(port_list),
        "uart_units_active": units_active,
        "uart_collector_active": bool(collector_active),
        "uart_bridge_active": bool(bridge_active),
        "uart_allow_share": os.environ.get("ALLOW_UART_SHARE", "0").strip() == "1",
        "uart_own_uartd": bool(own_uartd),
    }


def format_uart_contention_report(probe: dict[str, Any]) -> str:
    lines = [
        "UART contention check:",
        f"  ports={probe.get('uart_ports_checked')}",
        f"  contention={probe.get('uart_contention')} reasons={probe.get('uart_contention_reasons')}",
        f"  units_active={probe.get('uart_units_active')}",
    ]
    holders = probe.get("uart_port_holders") or []
    if not holders:
        lines.append("  holders=(none)")
    for h in holders:
        lines.append(
            f"  holder pid={h.get('pid')} port={h.get('port')} cmd={h.get('cmdline')!r}"
        )
    if probe.get("uart_collector_active") and not probe.get("uart_allow_share"):
        lines.append(
            "  FIX: sudo systemctl stop ethergold-collector.service  "
            "# prod listen steals ttyXRUSB RX from tip/witness"
        )
        lines.append(
            "  or permanent witness lock: touch /var/lib/ethergold/WITNESS_NO_PROD"
        )
    if probe.get("uart_bridge_active"):
        lines.append(
            "  FIX: sudo systemctl stop ethergold.service fixture-prod.service"
        )
    return "\n".join(lines)


def xr_get_reg(fd: int, reg: int) -> int:
    buf = struct.pack("iii", -1, reg, 0)
    buf = fcntl.ioctl(fd, XR_USB_SERIAL_GET_REG, buf)
    return struct.unpack("iii", buf)[2]


def xr_get_reg_ch(fd: int, channel: int, reg: int) -> int:
    buf = struct.pack("iii", channel, reg, 0)
    buf = fcntl.ioctl(fd, XR_USB_SERIAL_GET_REG, buf)
    return struct.unpack("iii", buf)[2]


def xr_b142x_channel(channel: BusChannel) -> int:
    """B142x block index: ch0=MST/OIS ttyXRUSB0, ch2=SLV/machine ttyXRUSB1."""
    return 2 if channel.key == "machine" else 0


def read_xr_driver_version_loaded(module: str = XR_DRIVER_MODULE) -> Optional[str]:
    """Version of the module currently in kernel memory (sysfs only)."""
    ver_path = Path(f"/sys/module/{module}/version")
    if not ver_path.is_file():
        return None
    text = ver_path.read_text(encoding="utf-8", errors="replace").strip()
    return text or None


def read_xr_driver_version_installed(module: str = XR_DRIVER_MODULE) -> Optional[str]:
    """Version of the on-disk .ko that modprobe would load next (modinfo)."""
    try:
        import subprocess

        out = subprocess.run(
            ["modinfo", "-F", "version", module],
            capture_output=True,
            text=True,
            timeout=5,
            check=False,
        )
        text = (out.stdout or "").strip()
        return text or None
    except Exception:
        return None


def read_xr_driver_version(module: str = XR_DRIVER_MODULE) -> Optional[str]:
    """Loaded kernel module version (sysfs). Falls back to modinfo only if not loaded."""
    loaded = read_xr_driver_version_loaded(module)
    if loaded is not None:
        return loaded
    return read_xr_driver_version_installed(module)


def parse_eg_version(text: str | None) -> tuple[int, ...] | None:
    """Parse ``A.B.C.D`` (or shorter) product versions. Returns None for legacy letter IDs."""
    if not text:
        return None
    s = str(text).strip()
    if not s or not s[0].isdigit():
        return None
    parts: list[int] = []
    for tok in s.split("."):
        if not tok.isdigit():
            return None
        parts.append(int(tok))
    if not parts:
        return None
    while len(parts) < 4:
        parts.append(0)
    return tuple(parts[:4])


def eg_version_cmp(a: str | None, b: str | None) -> int | None:
    """Compare two EG versions: -1 / 0 / 1, or None if either is not A.B.C.D."""
    pa, pb = parse_eg_version(a), parse_eg_version(b)
    if pa is None or pb is None:
        return None
    if pa < pb:
        return -1
    if pa > pb:
        return 1
    return 0


def eg_version_in_range(
    ver: str | None,
    *,
    min_ver: str = EG_COMPONENT_MIN_VERSION,
    max_ver: str = EG_COMPONENT_MAX_VERSION,
    legacy_ok: frozenset[str] = ACCEPTABLE_XR_DRIVER_LEGACY,
) -> bool:
    """True when ver is within [min, max], or is an allow-listed legacy ID."""
    if ver is None:
        return False
    if ver in legacy_ok:
        return True
    parsed = parse_eg_version(ver)
    if parsed is None:
        return False
    lo = parse_eg_version(min_ver)
    hi = parse_eg_version(max_ver)
    if lo is None or hi is None:
        return False
    return lo <= parsed <= hi


def probe_component_versions(
    *,
    uartd: Any | None = None,
    use_uartd: bool = False,
    use_uartd_inject: bool = False,
) -> dict[str, Any]:
    """Dropbox-visible stack versions for session_start — never SSH to ask."""
    loaded = read_xr_driver_version_loaded()
    installed = read_xr_driver_version_installed()
    # driver_version = what inject actually hits (loaded). Never prefer modinfo when loaded.
    driver_ver = loaded if loaded is not None else installed
    mismatch = (
        loaded is not None
        and installed is not None
        and loaded != installed
    )
    driver_ok = eg_version_in_range(driver_ver)
    out: dict[str, Any] = {
        "eg_version": EG_VERSION,
        "eg_version_slug": EG_VERSION_SLUG,
        "eg_version_expected": EG_VERSION,
        "eg_component_min": EG_COMPONENT_MIN_VERSION,
        "eg_component_max": EG_COMPONENT_MAX_VERSION,
        "driver_module": XR_DRIVER_MODULE,
        "driver_version": driver_ver,
        "driver_version_loaded": loaded,
        "driver_version_installed": installed,
        "driver_version_mismatch": bool(mismatch),
        "driver_version_expected": EXPECTED_XR_DRIVER_VERSION,
        "driver_version_ok": driver_ok,
        "driver_version_acceptable": driver_ok,
        "uartd_version_expected": EXPECTED_UARTD_VERSION,
        "inject_version_expected": EXPECTED_INJECT_VERSION,
        "parser_build_expected": FIXTURE_PARSER_BUILD,
    }
    # Client-side constants (what Python was built/synced with).
    try:
        from pog422_uart_client import (  # local import — optional on listen-only hosts
            POG422_INJECT_VERSION as _client_inject_ver,
            POG422_UARTD_VERSION as _client_uartd_ver,
        )

        out["uartd_version_client"] = _client_uartd_ver
        out["inject_version_client"] = _client_inject_ver
    except Exception:
        out["uartd_version_client"] = None
        out["inject_version_client"] = None

    out["uartd_version_daemon"] = None
    out["inject_version_daemon"] = None
    if uartd is not None and use_uartd:
        try:
            vers = uartd.get_versions()
            out["uartd_version_daemon"] = vers.get("uartd")
            out["inject_version_daemon"] = vers.get("inject")
            if vers.get("eg"):
                out["eg_version_daemon"] = vers.get("eg")
            if "hw_debug" in vers:
                out["hw_debug"] = int(vers.get("hw_debug") or 0)
        except Exception as exc:
            out["uartd_version_probe"] = str(exc)

    if "hw_debug" not in out:
        out["hw_debug"] = 0

    uartd_live = out.get("uartd_version_daemon") or out.get("uartd_version_client")
    inject_live = out.get("inject_version_daemon") or out.get("inject_version_client")
    out["uartd_version"] = uartd_live  # back-compat alias used by older evals
    out["inject_version"] = inject_live
    # uartd/inject must match Python tip — range lag caused silent pos5 misses.
    out["uartd_version_ok"] = (not use_uartd) or (uartd_live == EXPECTED_UARTD_VERSION)
    out["inject_version_ok"] = (not use_uartd_inject) or (
        inject_live == EXPECTED_INJECT_VERSION
    )
    eg_live = out.get("eg_version_daemon") or uartd_live or EG_VERSION
    out["eg_version_ok"] = eg_live == EG_VERSION
    # Stale RAM module (modinfo newer than sysfs) fails the stack even if in-range.
    out["component_versions_ok"] = bool(
        out["driver_version_ok"]
        and out["uartd_version_ok"]
        and out["inject_version_ok"]
        and out["eg_version_ok"]
        and not mismatch
    )
    return out


def probe_hat_session(
    port: str,
    *,
    ser: Any | None = None,
    uartd: Any | None = None,
    use_uartd: bool = False,
    use_uartd_inject: bool = False,
) -> dict[str, Any]:
    """Hat/driver facts for session_start (Dropbox-visible without SSH)."""
    del port, ser  # version probe does not need the fd when sysfs is available
    out = probe_component_versions(
        uartd=uartd, use_uartd=use_uartd, use_uartd_inject=use_uartd_inject
    )
    if uartd is not None:
        try:
            hr = uartd.hat_regs()
            if hr.err:
                out["hat_reg_probe"] = f"err={hr.err}"
            else:
                out["gpio_mode"] = hr.gpio_mode
                out["gpio_mode_hex"] = f"0x{hr.gpio_mode:04X}"
                out["uart_flow_reg"] = hr.uart_flow
                out["gpio_mode_manual_slv"] = hr.gpio_mode_manual_slv
                out["gpio_mode_ch0"] = hr.gpio_mode_ch0
                out["gpio_mode_ch0_hex"] = f"0x{hr.gpio_mode_ch0:04X}"
                out["uart_flow_ch0"] = hr.uart_flow_ch0
                out["gpio_mode_rs485_auto_ch0"] = hr.gpio_mode_rs485_auto_ch0
        except Exception as exc:
            out["hat_reg_probe"] = str(exc)
    return out


RX_FIFO_COUNT_REG = 0x044
TX_FIFO_COUNT_REG = 0x041
XR_USB_SERIAL_GET_RX_STATS = 0x80187610
FIONREAD = 0x541B
_RX_STATS_STRUCT = struct.Struct("QQII")


def uart_drain_no_usb(ser: serial.Serial, max_n: int = 8192) -> bytes:
    """Kernel tty drain. Never ``in_waiting`` / GET_REG (those USB-blocked ~2 s)."""
    try:
        fd = ser.fileno()
    except (OSError, AttributeError, TypeError):
        n = int(getattr(ser, "in_waiting", 0) or 0)
        if n <= 0:
            return b""
        try:
            return ser.read(min(n, max_n)) or b""
        except OSError:
            return b""
    from superball_slave_safety import tty_drain_available

    return tty_drain_available(int(fd), max_n=max_n)


def uart_rx_snapshot(ser: serial.Serial, channel: BusChannel) -> dict[str, Any]:
    """Kernel/chip/driver RX view — HS can toggle while all of these stay zero."""
    fd = ser.fileno()
    b142x = xr_b142x_channel(channel)
    snap: dict[str, Any] = {"b142x_ch": b142x}
    try:
        import array

        buf = array.array("i", [0])
        fcntl.ioctl(fd, FIONREAD, buf)
        snap["fionread"] = int(buf[0])
    except OSError as exc:
        snap["fionread_err"] = str(exc)
    snap["in_waiting"] = int(ser.in_waiting)
    try:
        snap["hw_fifo"] = int(xr_get_reg_ch(fd, b142x, RX_FIFO_COUNT_REG)) & 0x3FF
    except OSError as exc:
        snap["hw_fifo_err"] = str(exc)
    try:
        raw = fcntl.ioctl(fd, XR_USB_SERIAL_GET_RX_STATS, b"\0" * _RX_STATS_STRUCT.size)
        inserted, urbs, extra, pf = _RX_STATS_STRUCT.unpack(raw)
        snap["driver_rx_chars"] = int(inserted)
        snap["driver_urbs"] = int(urbs)
        snap["driver_extra_byte"] = int(extra)
        snap["driver_preciseflags"] = int(pf)
    except OSError as exc:
        snap["driver_rx_err"] = str(exc)
    return snap


def uart_driver_rx_delta(
    snap_now: dict[str, Any],
    snap_before: dict[str, Any] | None,
) -> int | None:
    if snap_before is None:
        return None
    before = snap_before.get("driver_rx_chars")
    now = snap_now.get("driver_rx_chars")
    if before is None or now is None:
        return None
    return int(now) - int(before)


def xr_set_reg(fd: int, reg: int, value: int) -> None:
    buf = struct.pack("iii", -1, reg, value)
    fcntl.ioctl(fd, XR_USB_SERIAL_SET_REG, buf)


def set_tx(ser: serial.Serial) -> None:
    fcntl.ioctl(ser.fileno(), XR_USB_SERIAL_SET_GPIO7, struct.pack("i", 1))


def set_rx(ser: serial.Serial) -> None:
    fcntl.ioctl(ser.fileno(), XR_USB_SERIAL_SET_GPIO7, struct.pack("i", 0))


class HandshakeMonitor:
    """GPIOx1 handshake input. POG422 hat is active-low (line low = bus driven)."""

    def __init__(self, ser: serial.Serial, channel: BusChannel, *, active_low: bool = True):
        self.channel = channel
        self.active_low = active_low
        self._fd = ser.fileno()
        # Keep/restore manual GPIO7 DE (same as uartd inject). Do not write MODE=0 —
        # that breaks python set_tx/set_rx (HS-only / Master never decodes).
        xr_set_reg(self._fd, GPIO_MODE_REG, POG422_GPIO_MODE_MANUAL_SLV)
        xr_set_reg(self._fd, GPIO_DIR_REG, POG422_GPIO7_DIR)

    def is_active(self) -> bool:
        status = xr_get_reg(self._fd, GPIO_STAT_REG)
        bit_set = bool(status & GPIO1_STAT_MASK)
        return not bit_set if self.active_low else bit_set


def slv_in_active(handshake: HandshakeMonitor) -> bool:
    """Read SLV-IN (GPIOB1): True when SLV-HS is active on the bus."""
    return handshake.is_active()


def slv_in_active_cached(handshake: HandshakeMonitor) -> bool:
    """Best-effort SLV-IN without a blocking IPC round-trip (log/diagnostics)."""
    cached = getattr(handshake, "is_active_cached", None)
    if callable(cached):
        return bool(cached())
    client = getattr(handshake, "client", None)
    if client is not None and hasattr(client, "slv_in_cached"):
        return bool(client.slv_in_cached())
    try:
        return handshake.is_active()
    except (TimeoutError, OSError, ConnectionError, RuntimeError):
        return False


def open_serial(
    port: str,
    channel: BusChannel,
    *,
    active_low: bool = True,
    hs_source: str = "uart",
) -> tuple[serial.Serial, HandshakeMonitor]:
    ser = serial.Serial(
        port,
        UART_BAUD,
        bytesize=serial.EIGHTBITS,
        parity=serial.PARITY_SPACE,
        stopbits=serial.STOPBITS_ONE,
        timeout=0.1,
    )
    set_rx(ser)
    # Pi HS: J3/J4 PI — XR GPIO1 is disconnected. Sense HS on BCM8/9; keep XR UART RX.
    source = (hs_source or "uart").strip().lower()
    if source in ("pi", "gpio", "j19", "pi_hs"):
        from pog422_gpio_hs import GpioHandshakeMonitor

        return ser, GpioHandshakeMonitor(channel.key, active_low=active_low)  # type: ignore[return-value]
    return ser, HandshakeMonitor(ser, channel, active_low=active_low)


def tid_le(data: bytes, offset: int = 3) -> Optional[str]:
    """Terminal ID in T340+ / topology format (wire bytes offset, offset+1)."""
    if offset + 1 >= len(data):
        return None
    return f"{data[offset]:02X}{data[offset + 1]:02X}"


def wire_bank_number(frame: bytes) -> Optional[int]:
    """Bank Number at wire byte 2 (after SOH + opcode). None if frame too short."""
    if len(frame) < 3 or frame[0] != SOH:
        return None
    return int(frame[POG422_BANK_NUMBER_OFFSET])


def resolve_local_bank_number() -> int:
    """Cabinet bank for stamp/filter. 0 = unset → accept all banks."""
    for key in ("EDGE_BANK", "FIELD_BANK", "POG422_BANK", "BANK_NUMBER"):
        raw = (os.environ.get(key) or "").strip()
        if raw.isdigit():
            value = int(raw)
            if value > 0:
                return value
    try:
        from cabinet_identity import latch_cabinet_identity

        bank = int(latch_cabinet_identity().bank_number or 0)
        if bank > 0:
            return bank
    except Exception:
        pass
    return 0


def local_bank_accepts(
    frame_bank: Optional[int],
    *,
    local_bank: Optional[int] = None,
) -> bool:
    """True when local bank is unset or matches frame bank byte."""
    lb = resolve_local_bank_number() if local_bank is None else int(local_bank)
    if lb <= 0:
        return True
    if frame_bank is None:
        return False
    return int(frame_bank) == lb


def packet_bank_event_fields(
    meta_or_bank: "PacketMeta | int | None",
    *,
    local_bank: Optional[int] = None,
) -> dict[str, Any]:
    """Stamp bank (+ local_bank / bank_ok) on every decode/capture event."""
    lb = resolve_local_bank_number() if local_bank is None else int(local_bank)
    if isinstance(meta_or_bank, PacketMeta):
        bank = meta_or_bank.bank_number
    else:
        bank = meta_or_bank
    fields: dict[str, Any] = {"bank": bank}
    if lb > 0:
        fields["local_bank"] = lb
        fields["bank_ok"] = local_bank_accepts(bank, local_bank=lb)
    else:
        fields["local_bank"] = None
        fields["bank_ok"] = True
    return fields


def parse_packet_meta(raw: bytes, *, index_in_window: int = 0, count_in_window: int = 1) -> PacketMeta:
    meta = PacketMeta(
        raw=raw,
        index_in_window=index_in_window,
        packet_count_in_window=count_in_window,
    )
    if len(raw) < 2 or raw[0] != SOH or raw[-1] != EOP:
        return meta
    meta.msg_type = raw[1]
    if len(raw) >= 3:
        meta.bank_number = int(raw[POG422_BANK_NUMBER_OFFSET])
    if meta.msg_type == 0x0D:
        if len(raw) >= 7:
            meta.game_id_hex = f"{raw[6]:02X}{raw[5]:02X}"
        if len(raw) >= 9:
            meta.session_counter = raw[8]
        if len(raw) >= 11:
            meta.master_tid_hex = tid_le(raw, 9)
    else:
        meta.tid_hex = tid_le(raw)
        if len(raw) >= 7:
            meta.game_id_hex = f"{raw[6]:02X}{raw[5]:02X}"
        if len(raw) >= 8:
            meta.play_sequence = raw[7]
    return meta


def status_close_dedupe_key(meta: PacketMeta) -> tuple[str, int] | None:
    """Key for gameplay Master status close dedupe (msg 0x0D, 14 B wire)."""
    if meta.msg_type != 0x0D:
        return None
    if not meta.game_id_hex or meta.session_counter is None:
        return None
    return (meta.game_id_hex, meta.session_counter)


STATUS_0D_WIRE_LEN = COMMAND_WIRE_LEN[0x0D]

_POG422_CHECKSUM_SELFTEST_LINES = (
    "05,0D,01,00,00,0C,97,00,8E,A4,5B,6E,DB,0D",
    "05,0D,01,00,00,0C,97,00,90,A4,5B,36,B9,0D",
    "05,81,01,38,D7,0C,97,00,00,AD,EE,0D",
    "05,21,01,1D,AF,07,91,00,00,EE,1A,0D",
)


def pog422_crc16_ccitt(data: bytes, init: int = 0xFFFF) -> int:
    """CRC-16-CCITT (poly 0x1021, MSB-first) used on POG422 SOH..EOP frames."""
    crc = init
    for b in data:
        crc ^= b << 8
        for _ in range(8):
            if crc & 0x8000:
                crc = ((crc << 1) ^ 0x1021) & 0xFFFF
            else:
                crc = (crc << 1) & 0xFFFF
    return crc


def pog422_set_frame_checksum(frame: bytearray) -> None:
    """Write big-endian CRC-16 into the two bytes before trailing EOP."""
    if len(frame) < 4 or frame[0] != SOH or frame[-1] != EOP:
        raise ValueError("invalid POG422 frame for checksum")
    crc = pog422_crc16_ccitt(bytes(frame[0:-3]))
    frame[-3] = (crc >> 8) & 0xFF
    frame[-2] = crc & 0xFF


def pog422_frame_checksum_ok(frame: bytes) -> bool:
    if len(frame) < 4 or frame[0] != SOH or frame[-1] != EOP:
        return False
    crc = pog422_crc16_ccitt(frame[0:-3])
    return crc == ((frame[-3] << 8) | frame[-2])


SCR_MSG = 0x03
SCR_REPLY = 0x0B

# Slave STOM comm requests (config / partial / SCR) — must inject to Master UART
# immediately; never game_transaction_gate defer (pos3 Local Net Test / stom_decoder).
STOM_COMM_REQUEST_MSGS = frozenset({0x01, 0x02, 0x04, SCR_MSG})

# STOM transaction path: requests + Master replies. 0x0A is stom_update/POG_UPDT
# (82 B) — same opcode as the periodic meter poll. EDGE spokes must inject it;
# copper already has it on the wire. Never game_transaction_gate / 0a_latest_wins.
STOM_COMM_PRIORITY_MSGS = frozenset(
    {0x01, 0x02, 0x04, SCR_MSG, SCR_REPLY, 0x09, 0x0A, 0x0E}
)


def is_stom_comm_request(msg_type: int | None) -> bool:
    return msg_type is not None and msg_type in STOM_COMM_REQUEST_MSGS


def is_stom_comm_priority(msg_type: int | None) -> bool:
    return msg_type is not None and msg_type in STOM_COMM_PRIORITY_MSGS


def eth_only_ports_unmapped(
    msg_type: int | None,
    *,
    topo: str,
) -> list[str] | None:
    """only_ports when TID has no gretap map. None = flood; [] = nowhere.

    Hub unknown cash/skill must not multi-spoke flood. T340+ CGS / Transfer
    Master Config (STOM) is undocumented-to-EG but copper-legal: the Master
    must see the request and the requester must see 0x09/0x0E. Never [].

    EG 4.1.0.1: SCR ``0x03``/``0x0B`` is **not** a bank broadcast. Hub unmapped
    SCR goes nowhere (``[]``) — never multi-spoke flood. Spoke unmapped SCR still
    floods the uplink so Local Net Test can reach Master.

    Spoke unmapped (non-SCR) always floods the uplink (Master path).
    """
    if is_scr_msg(msg_type):
        if (topo or "").strip().lower() == "spoke":
            return None
        return []
    if is_stom_comm_priority(msg_type):
        return None
    if (topo or "").strip().lower() == "spoke":
        return None
    return []


def inject_priority_class(
    msg_type: int | None,
    *,
    source: str = "eth_to_uart",
    tid_hex: str | None = None,
    bank_jpot_x1_armed_until_ns: dict[str, int] | None = None,
    now_ns: int | None = None,
) -> int:
    """Numeric eth→UART priority class for queue placement (higher first)."""
    if source == "status_0d_synth":
        return INJECT_PRIORITY_P0_CLOSE
    if is_bank_jpot_x1_priority_ack(
        msg_type,
        tid_hex,
        bank_jpot_x1_armed_until_ns,
        now_ns=now_ns,
    ):
        return INJECT_PRIORITY_BANK_JPOT_X1
    if is_scr_msg(msg_type):
        return INJECT_PRIORITY_SCR
    if is_stom_comm_priority(msg_type):
        return INJECT_PRIORITY_STOM
    return INJECT_PRIORITY_NORMAL


def inject_echo_bytes_match(sent: bytes, rx: bytes) -> tuple[bool, str]:
    """Score bus RX vs inject TX.

    Returns ``(ok, kind)`` where kind is ``exact``, ``prefix`` (self then peer),
    ``embedded`` (self found mid-buffer — peer may lead or trail), ``mismatch``,
    or ``none``.
    """
    if not rx or not sent:
        return False, "none"
    if rx == sent:
        return True, "exact"
    if len(rx) > len(sent) and rx.startswith(sent):
        return True, "prefix"
    # EG 2.2.2.69: peer-before-self (61|x9) is a successful self delivery with
    # leading peer — not a hard mismatch for harvest.
    if len(rx) > len(sent) and rx.find(sent) >= 0:
        return True, "embedded"
    return False, "mismatch"


def peer_uart_bytes_after_inject(sent: bytes, echo: bytes, post_de: bytes) -> bytes:
    """POG422-origin bytes in inject RX after stripping self-echo of *sent*.

    Inject-window RX is logged as inject_echo and cleared by uartd — peer/Master
    TxD glued into that buffer must still UART→ETH. Pure self-echo is not forwarded.

    Bidirectional peel (EG 2.2.2.69): when *sent* is embedded mid-buffer, keep
    **prefix before** and **suffix after** (pos5 18:33 — leading local ``61`` then
    self ``69``). Trailing-only peel discarded the ACK and forced infer.
    """
    echo = echo or b""
    post_de = post_de or b""
    sent = sent or b""

    def _peer_around_sent(bus: bytes) -> bytes:
        if not bus:
            return b""
        if not sent:
            return bus
        if bus == sent:
            return b""
        if bus.startswith(sent):
            return bus[len(sent) :]
        idx = bus.find(sent)
        if idx >= 0:
            return bus[:idx] + bus[idx + len(sent) :]
        return bus

    if echo:
        peer = _peer_around_sent(echo)
        if post_de:
            # Append post-DE only when it adds bytes not already present after peel.
            if peer and post_de == peer:
                return peer
            if peer and post_de.startswith(peer):
                return post_de
            if not peer:
                return _peer_around_sent(post_de)
            if post_de not in echo and not echo.endswith(post_de):
                return peer + post_de
        return peer
    return _peer_around_sent(post_de)


def is_scr_msg(msg_type: int | None) -> bool:
    """Send Com Request / reply (operator Local Net Tests menu only — not CGS)."""
    return msg_type in (SCR_MSG, SCR_REPLY)


def pog422_scr_diag_fields(pkt: bytes, meta: PacketMeta) -> dict[str, Any]:
    """Diagnostic fields for SCR (0x03) and enroll reply (0x0B) frames."""
    fields: dict[str, Any] = {
        "frame_hex": pkt.hex(),
        "wire_len": len(pkt),
        "soh_ok": len(pkt) >= 1 and pkt[0] == SOH,
        "eop_ok": len(pkt) >= 1 and pkt[-1] == EOP,
        "checksum_ok": pog422_frame_checksum_ok(pkt),
    }
    fields.update(packet_bank_event_fields(meta))
    if meta.msg_type is not None:
        known = wire_lengths_for_cmd(meta.msg_type)
        fields["wire_len_ok"] = len(pkt) in known if known else False
        fields["msg"] = meta.msg_type_hex
    if meta.tid_hex:
        fields["tid"] = meta.tid_hex
    if meta.game_id_hex:
        fields["game"] = meta.game_id_hex
    if meta.play_sequence is not None:
        fields["seq"] = meta.play_sequence
    if meta.master_tid_hex:
        fields["master_tid"] = meta.master_tid_hex
    return fields


def build_status_0d_from_template(template: bytes, counter: int) -> bytes | None:
    """Build a 14-byte Master status close from a cached template + session counter."""
    if len(template) != STATUS_0D_WIRE_LEN or template[0] != SOH or template[1] != 0x0D:
        return None
    frame = bytearray(template)
    frame[8] = counter & 0xFF
    pog422_set_frame_checksum(frame)
    return bytes(frame)


def build_status_0d_minimal(
    game_id_hex: str,
    counter: int,
    *,
    tid_tail: bytes | None = None,
    bank_number: int | None = None,
) -> bytes | None:
    """Experimental cold-start 0x0D: fixed prefix + game id + counter + optional TID tail."""
    if len(game_id_hex) != 4:
        return None
    game = int(game_id_hex, 16)
    bank = int(bank_number) if bank_number is not None else resolve_local_bank_number()
    if bank <= 0:
        bank = 1
    frame = bytearray(
        (
            SOH,
            0x0D,
            bank & 0xFF,
            0x00,
            0x00,
            game & 0xFF,
            (game >> 8) & 0xFF,
            0x00,
            counter & 0xFF,
            0x00,
            0x00,
            0x00,
            0x00,
            EOP,
        )
    )
    if tid_tail is not None and len(tid_tail) >= 2:
        frame[9] = tid_tail[0]
        frame[10] = tid_tail[1]
    pog422_set_frame_checksum(frame)
    return bytes(frame)


def status_0d_close_counter_from_x9(
    outcome: bytes,
    *,
    last_global: int | None,
    loose_counter: bool = False,
) -> tuple[int, str]:
    """Derive the Master 0x0D session counter for this round from the x9 outcome."""
    if loose_counter:
        return 0, "loose_zero"
    if len(outcome) > STATUS_0D_X9_COUNTER_BYTE:
        return outcome[STATUS_0D_X9_COUNTER_BYTE] & 0xFF, "x9_byte17"
    if last_global is not None:
        return (last_global + 1) & 0xFF, "global_next"
    return 1, "default_one"


def status_0d_close_counter_skill_grant(
    last_global: int | None,
) -> tuple[int, str]:
    """Second status close in a deal+skill round — one past the deal-path synth counter."""
    if last_global is not None:
        return (last_global + 1) & 0xFF, "skill_grant_next"
    return 1, "default_one"


def status_0d_counter_hint_from_x9(frame: bytes, play_sequence: int | None) -> int:
    """Legacy alias — prefer status_0d_close_counter_from_x9()."""
    counter, _ = status_0d_close_counter_from_x9(frame, last_global=None)
    return counter


def status_0d_bank_tid_tail(
    wager: bytes | None,
    outcome: bytes,
    *,
    progressive_tail: bytes | None = None,
) -> bytes:
    """Bytes 9-10 of the 14-byte Master status close (LE master entity tail).

    Copper: constant progressive master id from 0x0A (e.g. A4,5B). Wager/x9 carry
    slave tid at bytes 3-4 and game at 5-6; counter at outcome byte 17.
    """
    if progressive_tail is not None and len(progressive_tail) >= 2:
        return bytes(progressive_tail[:2])
    return STATUS_0D_DEFAULT_BANK_TID_TAIL


def build_status_0d_from_round(
    wager: bytes | None,
    outcome: bytes,
    *,
    counter: int,
    progressive_tail: bytes | None = None,
) -> bytes | None:
    """Build Master status close from slave wager (x0) + Master outcome (x9)."""
    if len(outcome) < 7:
        return None
    game_id_hex = f"{outcome[6]:02X}{outcome[5]:02X}"
    tid_tail = status_0d_bank_tid_tail(wager, outcome, progressive_tail=progressive_tail)
    return build_status_0d_minimal(game_id_hex, counter, tid_tail=tid_tail)


def verify_pog422_checksum_selftest() -> None:
    for line in _POG422_CHECKSUM_SELFTEST_LINES:
        raw = bytes(int(x, 16) for x in line.split(","))
        if not pog422_frame_checksum_ok(raw):
            raise RuntimeError(f"pog422 checksum selftest failed: {line}")
        patched = bytearray(raw)
        pog422_set_frame_checksum(patched)
        if bytes(patched) != raw:
            raise RuntimeError(f"pog422 checksum rewrite failed: {line}")
    minimal = build_status_0d_minimal("970C", 0x8E, tid_tail=bytes((0xA4, 0x5B)))
    if minimal is None or len(minimal) != STATUS_0D_WIRE_LEN:
        raise RuntimeError("build_status_0d_minimal failed")
    if not pog422_frame_checksum_ok(minimal):
        raise RuntimeError("build_status_0d_minimal checksum invalid")
    # v13 pos1 Touch 6 round 1 — wager + x29 → 0x0D (counter from x9 byte 17).
    wager = bytes(
        int(x, 16)
        for x in "05,20,01,1D,AF,07,91,06,04,00,00,00,00,00,00,00,00,61,BC,00,00,00,00,00,00,00,00,00,00,01,31,41,0E,C3,40,00,0D,9A,50,00,0C,13,05,08,0C,C1,6B,0D".split(
            ","
        )
    )
    outcome = bytes(
        int(x, 16)
        for x in "05,29,01,1D,AF,07,91,06,00,00,07,A2,8E,00,00,00,00,FE,00,6E,F0,04,0F,13,05,07,15,1A,38,0D".split(
            ","
        )
    )
    expect = bytes(
        int(x, 16)
        for x in "05,0D,01,00,00,07,91,00,FE,A4,5B,B8,F4,0D".split(",")
    )
    built = build_status_0d_from_round(
        wager,
        outcome,
        counter=0xFE,
        progressive_tail=STATUS_0D_DEFAULT_BANK_TID_TAIL,
    )
    if built != expect:
        raise RuntimeError(
            f"build_status_0d_from_round selftest failed: got {built.hex()} want {expect.hex()}"
        )
    counter, src = status_0d_close_counter_from_x9(outcome, last_global=None)
    if counter != 0xFE or src != "x9_byte17":
        raise RuntimeError("status_0d_close_counter_from_x9 selftest failed")
    # v15c — bank counter 255→0 wrap: byte 17 == 0 is valid, not "missing".
    wrap_outcome = bytearray(outcome)
    wrap_outcome[STATUS_0D_X9_COUNTER_BYTE] = 0
    counter, src = status_0d_close_counter_from_x9(
        bytes(wrap_outcome),
        last_global=255,
    )
    if counter != 0 or src != "x9_byte17":
        raise RuntimeError(
            f"x9 byte17 zero-wrap selftest failed: counter={counter} src={src}"
        )
    skill_counter, skill_src = status_0d_close_counter_skill_grant(181)
    if skill_counter != 182 or skill_src != "skill_grant_next":
        raise RuntimeError(
            f"skill grant close counter selftest failed: counter={skill_counter} src={skill_src}"
        )
    keno_template = bytes(
        int(x, 16)
        for x in (
            "05,49,01,A4,5B,09,93,07,00,00,07,A1,39,00,00,00,00,08,00,"
            "6E,FA,39,34,43,3C,4F,36,1E,41,3D,3B,47,33,2F,06,2B,4B,12,37,"
            "3F,3A,D2,96,0D"
        ).split(",")
    )
    keno_synth = build_keno_jpot_synth(keno_template, jpot_amount=500_088)
    if keno_synth is None:
        raise RuntimeError("build_keno_jpot_synth failed")
    if not pog422_frame_checksum_ok(keno_synth):
        raise RuntimeError("build_keno_jpot_synth checksum invalid")
    if keno_synth[KENO_RESULT_ACTIVEWINS_OFFSET] != 1:
        raise RuntimeError("build_keno_jpot_synth activewins not set")
    progwin = progressive_progwin_cents_be(keno_synth)
    if progwin != 500_088:
        raise RuntimeError(f"build_keno_jpot_synth progwin cents {progwin} != 500088")
    if keno_synth[PROGRESSIVE_PROGDSP_OFFSET : PROGRESSIVE_PROGDSP_OFFSET + 4] != bytes(
        (0x00, 0x07, 0xA1, 0x78)
    ):
        raise RuntimeError("build_keno_jpot_synth progdsp BE mismatch")
    if list(keno_synth[KENO_RESULT_BALL_OFFSET : KENO_RESULT_BALL_OFFSET + 20]) != list(
        keno_lab_default_wire_balls()
    ):
        raise RuntimeError("build_keno_jpot_synth balls mismatch")
    for i, n in enumerate(KENO_LAB_DEFAULT_DISPLAY_BALLS):
        got = keno_display_ball_from_wire(
            keno_synth[KENO_RESULT_BALL_OFFSET + i]
        )
        if got != n:
            raise RuntimeError(f"build_keno_jpot_synth display ball {i}: {got} != {n}")
    touch6_template = bytes(
        int(x, 16)
        for x in "05,29,01,1D,AF,07,91,06,00,00,07,A2,8E,00,00,00,00,FE,00,6E,F0,04,0F,13,05,07,15,1A,38,0D".split(
            ","
        )
    )
    touch6_wager_base = bytes(
        int(x, 16)
        for x in "05,20,01,1D,AF,07,91,06,04,00,00,00,00,00,00,00,00,61,BC,00,00,00,00,00,00,00,00,00,00,01,31,41,0E,C3,40,00,0D,9A,50,00,0C,13,05,08,0C,C1,6B,0D".split(
            ","
        )
    )

    def _touch6_wager_with_cards(*card_masks: int) -> bytes:
        w = bytearray(touch6_wager_base)
        for i, mask in enumerate(card_masks):
            off = TOUCH6_WAGER_CARD_OFFSET + i * 4
            w[off : off + 4] = int(mask).to_bytes(4, "little")
        pog422_set_frame_checksum(w)
        return bytes(w)

    card1_6of6 = TOUCH6_LOT6_PLAY_MASK | 0x3F
    card2_3of6 = TOUCH6_LOT6_PLAY_MASK | 0x07
    w1 = _touch6_wager_with_cards(card1_6of6, 0, 0, 0)
    s1 = build_touch6_result_synth(touch6_template, jpot_amount=500_088, wager=w1)
    if s1 is None or s1[PROGRESSIVE_ACTIVEWINS_OFFSET] != 1:
        raise RuntimeError("touch6 synth game1 activewins != 1")
    if list(s1[TOUCH6_RESULT_BALL_OFFSET : TOUCH6_RESULT_BALL_OFFSET + 6]) != list(
        TOUCH6_LAB_DEFAULT_BALLS
    ):
        raise RuntimeError("touch6 synth balls not 1-6")
    if progressive_progwin_cents_be(s1) != 500_088:
        raise RuntimeError("touch6 synth game1 progwin mismatch")
    w2 = _touch6_wager_with_cards(card1_6of6, card2_3of6, 0, 0)
    s2 = build_touch6_result_synth(touch6_template, wager=w2)
    if s2 is None or s2[PROGRESSIVE_ACTIVEWINS_OFFSET] != 1:
        raise RuntimeError("touch6 synth game2 activewins != 1 (card1 only jpot)")
    if progressive_progwin_cents_be(s2) == 0:
        raise RuntimeError("touch6 synth game2 progwin should be set")
    w3 = _touch6_wager_with_cards(card1_6of6, card1_6of6, 0, 0)
    s3 = build_touch6_result_synth(touch6_template, wager=w3)
    if s3 is None or s3[PROGRESSIVE_ACTIVEWINS_OFFSET] != 2:
        raise RuntimeError("touch6 synth game3 activewins != 2")
    w0 = _touch6_wager_with_cards(card2_3of6, 0, 0, 0)
    s0 = build_touch6_result_synth(touch6_template, wager=w0)
    if s0 is None or progressive_progwin_cents_be(s0) != 0:
        raise RuntimeError("touch6 synth 3/6 only must not set progwin")
    bingo_native_calls = (
        34, 41, 35, 57, 26, 23, 14, 75, 22, 43, 51, 56, 20, 72, 11,
        2, 28, 24, 17, 55, 25, 49, 7, 53, 5, 70, 50, 39, 1, 69,
        18, 40, 31, 8, 10, 52, 47, 21, 6, 63, 30, 48, 58, 54, 19,
        65, 61, 12, 4, 68,
    )
    bingo_template = bytearray(
        [
            0x05, 0x39, 0x01, 0x38, 0xD7, 0x06, 0x92, 0x07,
            0x00, 0x00, 0x06, 0x1A, 0x9F, 0x00, 0x00, 0x00, 0x00,
            0x99, 0x00, 0xF5, 0xF4,
            *bingo_native_calls,
            0x00, 0x00, 0x0D,
        ]
    )
    pog422_set_frame_checksum(bingo_template)
    bingo_template = bytes(bingo_template)
    if len(bingo_template) != 74:
        raise RuntimeError(f"bingo selftest template len {len(bingo_template)} != 74")
    bingo_synth = build_bingo_jpot_synth(bingo_template)
    if bingo_synth is None:
        raise RuntimeError("build_bingo_jpot_synth failed")
    if not pog422_frame_checksum_ok(bingo_synth):
        raise RuntimeError("build_bingo_jpot_synth checksum invalid")
    if bingo_synth[PROGRESSIVE_ACTIVEWINS_OFFSET] != 1:
        raise RuntimeError("build_bingo_jpot_synth activewins not set")
    if progressive_progwin_cents_be(bingo_synth) != BINGO_LAB_DEFAULT_JPOT_CENTS:
        raise RuntimeError("build_bingo_jpot_synth progwin mismatch")
    expected_calls = bingo_coverall_calls_from_grid(BINGO_LAB_SETA_CARD1_GRID)
    if bingo_calls_from_result(bingo_synth) != expected_calls:
        raise RuntimeError("build_bingo_jpot_synth calls mismatch")
    triple7s_template = bytes(
        int(x, 16)
        for x in "05,89,01,38,D7,0C,97,07,00,00,07,A6,56,00,00,00,00,9B,00,F5,F6,06,05,03,0B,03,01,02,07,01,03,04,0A,15,0A,0D".split(
            ","
        )
    )
    triple7s_synth = build_triple7s_jpot_synth(triple7s_template)
    if triple7s_synth is None:
        raise RuntimeError("build_triple7s_jpot_synth failed")
    if not pog422_frame_checksum_ok(triple7s_synth):
        raise RuntimeError("build_triple7s_jpot_synth checksum invalid")
    triple7s_grid = slot_grid_wire_from_result(triple7s_synth)
    if triple7s_grid is None or triple7s_grid[3:6] != (SLOT_WIRE_TRIPLE7,) * 3:
        raise RuntimeError("build_triple7s_jpot_synth payline1 not TRIPLE7")
    superpick_template = bytes(
        int(x, 16)
        for x in "05,69,01,1D,AF,07,95,06,00,00,07,A2,8E,00,00,00,00,FE,00,6E,F0,04,0F,13,05,07,15,1A,38,0D".split(
            ","
        )
    )
    superpick_wager_base = bytes(
        int(x, 16)
        for x in "05,60,01,1D,AF,07,95,06,04,00,00,00,00,00,00,00,00,61,BC,00,00,00,00,00,00,00,00,00,00,01,31,41,0E,C3,40,00,0D,9A,50,00,0C,13,05,08,0C,C1,6B,0D".split(
            ","
        )
    )

    def _superpick_wager_with_cards(*card_masks: int) -> bytes:
        w = bytearray(superpick_wager_base)
        for i, mask in enumerate(card_masks):
            off = SUPERPICK_WAGER_CARD_OFFSET + i * 4
            w[off : off + 4] = int(mask).to_bytes(4, "little")
        pog422_set_frame_checksum(w)
        return bytes(w)

    # Card 1: six picks of 0 match forced all-zero draw → FIRST 6 / progressive.
    card1_jpot = SUPERPICK_PLAY_MASK
    # Card 2: picks 0,0,0 at positions 0–2 only → 3 FIRST hits, no progressive.
    card2_3pick = SUPERPICK_PLAY_MASK | (9 << 12) | (9 << 16) | (9 << 20)
    w1 = _superpick_wager_with_cards(card1_jpot, 0, 0, 0)
    s1 = build_superpick_result_synth(superpick_template, jpot_amount=500_088, wager=w1)
    if s1 is None or s1[PROGRESSIVE_ACTIVEWINS_OFFSET] != 1:
        raise RuntimeError("superpick synth game1 activewins != 1")
    if list(s1[SUPERPICK_RESULT_BALL_OFFSET : SUPERPICK_RESULT_BALL_OFFSET + 6]) != list(
        SUPERPICK_LAB_DEFAULT_BALLS
    ):
        raise RuntimeError("superpick synth balls not all zero")
    w2 = _superpick_wager_with_cards(card1_jpot, card2_3pick, 0, 0)
    s2 = build_superpick_result_synth(superpick_template, wager=w2)
    if s2 is None or s2[PROGRESSIVE_ACTIVEWINS_OFFSET] != 1:
        raise RuntimeError("superpick synth game2 activewins != 1 (card1 only jpot)")
    w3 = _superpick_wager_with_cards(card1_jpot, card1_jpot, 0, 0)
    s3 = build_superpick_result_synth(superpick_template, wager=w3)
    if s3 is None or s3[PROGRESSIVE_ACTIVEWINS_OFFSET] != 2:
        raise RuntimeError("superpick synth game3 activewins != 2")
    w0 = _superpick_wager_with_cards(card2_3pick, 0, 0, 0)
    s0 = build_superpick_result_synth(superpick_template, wager=w0)
    if s0 is None or progressive_progwin_cents_be(s0) != 0:
        raise RuntimeError("superpick synth 3-pick only must not set progwin")
    poker_template = bytes(
        int(x, 16)
        for x in (
            "05,59,01,38,D7,00,94,03,00,00,01,F7,BA,00,00,00,00,74,00,F3,CF,"
            "32,26,11,2A,33,31,21,29,0C,13,0A,87,0D"
        ).split(",")
    )
    poker_synth = build_poker_royal_flush_synth(poker_template)
    if poker_synth is None:
        raise RuntimeError("build_poker_royal_flush_synth failed")
    if not pog422_frame_checksum_ok(poker_synth):
        raise RuntimeError("build_poker_royal_flush_synth checksum invalid")
    royal = POKER_LAB_ROYAL_FLUSH_HEARTS_WIRE
    deal = tuple(poker_synth[POKER_CARD_DEAL_START : POKER_CARD_DEAL_START + 5])
    if deal != royal:
        raise RuntimeError(f"build_poker_royal_flush_synth deal {deal}")
    wires = poker_card_wires_in_frame(poker_synth)
    if not poker_card_wires_unique(wires):
        raise RuntimeError("build_poker_royal_flush_synth must emit 10 unique cards")
    # Bench 2026-07-02 GDE capture: royal deal + Master's draw-pile, de-duped.
    poker_initial_deal = bytes(
        int(x, 16)
        for x in (
            "05,59,01,38,D7,00,94,07,00,00,01,F7,C2,00,00,00,00,78,00,F3,D3,"
            "05,33,17,12,0A,3B,37,0D,01,1B,08,08"
        ).split(",")
    )
    poker_initial_synth = build_poker_royal_flush_synth(poker_initial_deal)
    if poker_initial_synth is None:
        raise RuntimeError("build_poker_royal_flush_synth initial deal failed")
    if tuple(poker_initial_synth[21:26]) != royal:
        raise RuntimeError("initial deal synth must patch deal region")
    if not poker_card_wires_unique(poker_card_wires_in_frame(poker_initial_synth)):
        raise RuntimeError("initial deal synth must keep 10 unique cards")
    # K♥ and A♥ in Master draw-pile collide with royal — expect filler substitution.
    draw = tuple(poker_initial_synth[26:31])
    if draw == (0x3B, 0x37, 0x0D, 0x01, 0x1B):
        raise RuntimeError("initial deal synth must de-dupe draw-pile vs royal deal")
    spinjack_template = bytes(
        int(x, 16)
        for x in (
            "05,79,01,38,D7,0F,96,06,00,00,01,90,2D,00,00,00,00,C5,00,F4,20,"
            "0B,0C,03,3D,35,17,28,34,01,32,02,39,2B,26,3C,07,13,04,4B,0A,0D"
        ).split(",")
    )
    spinjack_synth = build_spinjack_flushjack_synth(spinjack_template)
    if spinjack_synth is None:
        raise RuntimeError("build_spinjack_flushjack_synth failed")
    if not pog422_frame_checksum_ok(spinjack_synth):
        raise RuntimeError("build_spinjack_flushjack_synth checksum invalid")
    if spinjack_synth[22] != SPINJACK_LAB_FLUSHJACK_HEARTS_WIRE[0]:
        raise RuntimeError("build_spinjack_flushjack_synth player ace byte 22")
    if spinjack_synth[24] != SPINJACK_LAB_FLUSHJACK_HEARTS_WIRE[1]:
        raise RuntimeError("build_spinjack_flushjack_synth player ten byte 24")
    if not spinjack_card_wires_unique(spinjack_card_wires_in_frame(spinjack_synth)):
        raise RuntimeError("build_spinjack_flushjack_synth must emit unique card bytes")


def eth_forward_ok(raw: bytes) -> tuple[bool, str | None]:
    """True when raw is a complete frame suitable for bridge forward/inject.

    Reject EOP-hunt crops: payload 0x0D is not packet boundary unless wire len matches.
    """
    if len(raw) < 2 or raw[0] != SOH or raw[-1] != EOP:
        return False, "bad_frame"
    if not wire_len_valid(raw):
        return False, "wire_len_crop"
    return True, None


# OIS wire: variable length, opaque payload — forward HS window bytes as-is.
OIS_RAW_MAX_BYTES = 4096


def ois_raw_forward_ok(raw: bytes) -> tuple[bool, str | None]:
    if len(raw) < 1:
        return False, "empty"
    if len(raw) > OIS_RAW_MAX_BYTES:
        return False, "too_long"
    return True, None


# Dual-channel bridge: one leading byte on 0x88B5 payload before UART bytes (fixture Pis only).
FIXTURE_ETH_TAG_MACHINE = 0x4D  # 'M' — ttyXRUSB1 / GPIOB1 gameplay bus
FIXTURE_ETH_TAG_OIS = 0x4F  # 'O' — ttyXRUSB0 / GPIOA1 OIS bus
FIXTURE_ETH_TAG_PIPE = 0x50  # 'P' — EDGE pipe probe (never UART-injected)
FIXTURE_ETH_TAG_TO_CHANNEL: dict[int, str] = {
    FIXTURE_ETH_TAG_MACHINE: "machine",
    FIXTURE_ETH_TAG_OIS: "ois",
    FIXTURE_ETH_TAG_PIPE: "pipe",
}


def fixture_eth_tag_payload(
    channel: BusChannel,
    payload: bytes,
    *,
    dual: bool,
) -> bytes:
    if not dual:
        return payload
    tag = FIXTURE_ETH_TAG_OIS if channel.key == "ois" else FIXTURE_ETH_TAG_MACHINE
    return bytes([tag]) + payload


def fixture_eth_untag_payload(
    wire: bytes,
    *,
    dual: bool,
) -> tuple[str | None, bytes, str | None]:
    """Return (channel_key, uart_payload, skip_reason)."""
    if not dual:
        return "machine", wire, None
    if len(wire) < 2:
        return None, b"", "dual_tag_missing"
    tag = wire[0]
    ch = FIXTURE_ETH_TAG_TO_CHANNEL.get(tag)
    if ch is None:
        return None, wire, "dual_tag_unknown"
    return ch, wire[1:], None


class UartOriginTracker:
    """Track frames this node recently forwarded UART→ETH.

    Not TID/opcode filtering. Only blocks re-injecting the Ethernet broadcast echo
    of a frame this same Pi just sent UART→ETH.

    Pos 3: Master/pos-2 bus traffic → UART → ETH is unaffected. Pos 4 Ethernet
    traffic is never in this list and always injects to the POG422 bus.
    """

    def __init__(self, *, hold_ms: float = 2000.0, max_entries: int = 256):
        self.hold_ns = int(hold_ms * 1_000_000)
        self.max_entries = max_entries
        self._entries: list[tuple[int, bytes]] = []

    def note_uart_to_eth(self, frame: bytes) -> None:
        now = time.perf_counter_ns()
        self._entries.append((now, frame))
        if len(self._entries) > self.max_entries:
            self._entries.pop(0)
        self._prune(now)

    def is_uart_origin_echo(self, frame: bytes) -> bool:
        now = time.perf_counter_ns()
        self._prune(now)
        return any(seen == frame for _, seen in self._entries)

    def _prune(self, now: int) -> None:
        cutoff = now - self.hold_ns
        self._entries = [(ts, frame) for ts, frame in self._entries if ts >= cutoff]


# Companion captures opened by a second process (e.g. hs_gpio_logger). Wiping
# them while that fd is open orphans the inode — console still counts edges,
# Dropbox sync never sees the file. Keep them across CLEAR_CAPTURE.
_CLEAR_CAPTURE_PRESERVE_NAMES = frozenset({"hs_gpio_machine.jsonl"})
_CLEAR_CAPTURE_PRESERVE_PREFIXES = ("hs_gpio_",)


def _is_capture_companion(entry: Path) -> bool:
    """Out-of-band instrument capture (hs_gpio logger), not written by the bridge."""
    name = entry.name
    if name in _CLEAR_CAPTURE_PRESERVE_NAMES:
        return True
    return any(name.startswith(prefix) for prefix in _CLEAR_CAPTURE_PRESERVE_PREFIXES)


def _capture_entry_in_use(entry: Path) -> bool:
    """True when a live process holds this exact inode open.

    Compares dev/inode instead of the /proc symlink text, because an fd whose path
    was already unlinked reads back as ``<path> (deleted)``.
    """
    try:
        target = entry.stat()
    except OSError:
        return False
    proc = Path("/proc")
    if not proc.is_dir():
        return False
    for pid_dir in proc.iterdir():
        if not pid_dir.name.isdigit():
            continue
        try:
            fds = list((pid_dir / "fd").iterdir())
        except OSError:
            continue  # process exited, or not ours to inspect
        for fd in fds:
            try:
                st = fd.stat()  # follows the link
            except OSError:
                continue
            if st.st_ino == target.st_ino and st.st_dev == target.st_dev:
                return True
    return False


def _is_operational_capture_log(entry: Path) -> bool:
    """pogcc-com.log lives in logs/ so sync_to_dropbox sees Heltec/P3 stalls.

    Always keep it (and rotations). It is not a listen session file.
    """
    return entry.name == "pogcc-com.log" or entry.name.startswith("pogcc-com.log.")


def _preserve_capture_entry(entry: Path) -> bool:
    """Keep companion captures while held open; always keep pogcc-com logs.

    A leftover hs_gpio file with no holder is stale from an earlier run. Keeping it
    would fold old edges into this session's archive, so the analyzer would
    correlate HS timing against the wrong events.jsonl — a silent wrong answer,
    which is worse than no data.
    """
    if _is_operational_capture_log(entry):
        return True
    return _is_capture_companion(entry) and _capture_entry_in_use(entry)


def clear_capture_files(
    output_dir: Path,
    *,
    logs_dir: Path | None = None,
) -> list[str]:
    """Wipe prior session capture dirs (data + optional logs) in place.

    Removes contents — including rotated ``events.jsonl.*`` /
    ``packets.raw.*`` — not only the active filenames. Directories themselves
    are kept. Launchers with CLEAR_CAPTURE=1 wipe before starting uartd, then
    pass ``--append-capture`` so this is not re-run over a live uartd.log.

    Preserves ``hs_gpio*`` companion files **that a live process holds open**, so a
    foreground GPIO logger started before the bridge is not unlinked out from under
    its fd. Stale companions with no holder are removed like anything else.
    Always keeps ``pogcc-com.log*`` (Heltec/P3 operational log for Dropbox sync).
    """
    removed: list[str] = []
    preserved: list[str] = []
    stale_companions: list[str] = []

    def _wipe(path: Path) -> None:
        path.mkdir(parents=True, exist_ok=True)
        for entry in list(path.iterdir()):
            if _preserve_capture_entry(entry):
                preserved.append(str(entry))
                continue
            if _is_capture_companion(entry):
                stale_companions.append(str(entry))
            try:
                if entry.is_dir() and not entry.is_symlink():
                    shutil.rmtree(entry)
                else:
                    entry.unlink(missing_ok=True)
                removed.append(str(entry))
            except OSError:
                pass

    _wipe(output_dir)
    if logs_dir is not None:
        _wipe(logs_dir)
    if preserved:
        print(
            "CLEAR_CAPTURE preserve (open by a live logger): " + ", ".join(preserved),
            flush=True,
        )
    if stale_companions:
        print(
            "CLEAR_CAPTURE removed stale companion capture: "
            + ", ".join(stale_companions),
            flush=True,
        )
    return removed


def split_packets(raw: bytes, *, mark_start: int | None = None) -> list[bytes]:
    """Split packets from a completed HS capture window.

    Only COMMAND_WIRE_LEN* lengths are accepted. Payload bytes may equal 0x0D;
    never hunt interior EOP — that produced 18/15 B false 0x79 crops on glued windows.
    """
    if mark_start is None:
        mark_start = SOH
    packets: list[bytes] = []
    i = 0
    while i < len(raw):
        if raw[i] != mark_start:
            i += 1
            continue
        if i + 1 >= len(raw):
            break
        cmd = raw[i + 1]
        known_lens = wire_lengths_for_cmd(cmd)
        framed = False
        for wire_len in known_lens:
            if i + wire_len > len(raw):
                continue
            pkt = raw[i : i + wire_len]
            if pkt[0] == mark_start and pkt[-1] == EOP:
                packets.append(bytes(pkt))
                i += wire_len
                framed = True
                break
        if framed:
            continue
        if known_lens:
            # Opcode is in the table but no valid length fit — skip false SOH sync.
            i += 1
            continue
        # Unknown opcode: accept only a tail segment ending at buffer end (no interior hunt).
        eop_idx = raw.find(EOP, i + 1)
        if eop_idx < 0 or eop_idx + 1 != len(raw):
            i += 1
            continue
        seg = raw[i : eop_idx + 1]
        if len(seg) >= 2 and seg[0] == mark_start:
            packets.append(bytes(seg))
        break
    return packets


def strip_before_mark(buffer: bytes, *, mark_start: int | None = None) -> bytes:
    if mark_start is None:
        mark_start = SOH
    mark_index = buffer.find(bytes([mark_start]))
    if mark_index < 0:
        return b""
    return buffer[mark_index:]


def normalize_window(buffer: bytes, *, mark_start: int | None = None) -> Optional[bytes]:
    """Trim leading SPACE bytes; return the full HS burst (GPIOB1-active window).

    Packet boundaries come from COMMAND_WIRE_LEN in split_packets(), not from
    hunting 0x0D inside the payload.
    """
    if not buffer:
        return None
    trimmed = strip_before_mark(buffer, mark_start=mark_start)
    if not trimmed:
        return None
    return trimmed


def uart_air_time_us(length: int) -> float:
    """9-bit UART time at 57600 baud."""
    return (length * 11 / UART_BAUD) * 1_000_000.0


@dataclass
class HsWitnessCounters:
    """Live UART/HS capture counters (console + fixture_stats.json)."""

    hs_inactive: int = 0
    hs_empty_bsy: int = 0
    hs_with_data: int = 0
    bytes_discarded_inactive: int = 0
    hs_fifo_diag: int = 0
    window_skip: int = 0
    packet_complete: int = 0
    packet_complete_by_msg: dict[str, int] = field(default_factory=dict)

    def as_dict(self) -> dict[str, Any]:
        return asdict(self)


def hs_porch_timing_fields(
    *,
    window_start_perf_ns: Optional[int],
    first_byte_perf_ns: Optional[int],
    last_byte_perf_ns: Optional[int],
    hs_inactive_perf_ns: Optional[int],
) -> dict[str, Any]:
    """HS rise → first byte → last byte → HS fall porches (µs).

    Requires honest first/last stamps (HW_DEBUG UARTD streams during HS).
    Prod burst-on-fall often collapses byte_span to 0.
    """
    out: dict[str, Any] = {}
    if window_start_perf_ns is not None:
        out["t_hs_rise_perf_ns"] = int(window_start_perf_ns)
    if first_byte_perf_ns is not None:
        out["t_first_byte_perf_ns"] = int(first_byte_perf_ns)
    if last_byte_perf_ns is not None:
        out["t_last_byte_perf_ns"] = int(last_byte_perf_ns)
    if hs_inactive_perf_ns is not None:
        out["t_hs_fall_perf_ns"] = int(hs_inactive_perf_ns)

    if (
        window_start_perf_ns is not None
        and first_byte_perf_ns is not None
        and first_byte_perf_ns >= window_start_perf_ns
    ):
        out["front_porch_us"] = round(
            (first_byte_perf_ns - window_start_perf_ns) / 1000.0, 1
        )
    if (
        first_byte_perf_ns is not None
        and last_byte_perf_ns is not None
        and last_byte_perf_ns >= first_byte_perf_ns
    ):
        out["byte_span_us"] = round(
            (last_byte_perf_ns - first_byte_perf_ns) / 1000.0, 1
        )
    if (
        last_byte_perf_ns is not None
        and hs_inactive_perf_ns is not None
        and hs_inactive_perf_ns >= last_byte_perf_ns
    ):
        out["back_porch_us"] = round(
            (hs_inactive_perf_ns - last_byte_perf_ns) / 1000.0, 1
        )
    if (
        window_start_perf_ns is not None
        and hs_inactive_perf_ns is not None
        and hs_inactive_perf_ns >= window_start_perf_ns
    ):
        out["hs_width_us"] = round(
            (hs_inactive_perf_ns - window_start_perf_ns) / 1000.0, 1
        )
    return out


class EventLog:
    def __init__(
        self,
        output_dir: Path,
        identity: FixtureIdentity,
        *,
        echo: bool = False,
        prod_mode: bool = False,
        dev_light: bool = False,
        hs_witness: bool = True,
    ):
        self.output_dir = output_dir
        self.identity = identity
        self.echo = echo
        self.prod_mode = prod_mode
        self.dev_light = bool(dev_light) and not prod_mode
        self.hs_witness = bool(hs_witness) and not prod_mode and not self.dev_light
        self.witness = HsWitnessCounters()
        events_name = PROD_ERRORS_JSONL if prod_mode else EVENTS_JSONL
        self.events_path = output_dir / events_name
        self.packets_path = output_dir / PACKETS_RAW
        self.inject_echo_path = output_dir / INJECT_ECHO_JSONL
        self.hw_debug_path = output_dir / HW_DEBUG_JSONL
        self.stats_path = output_dir / FIXTURE_STATS
        self._events = self.events_path.open("a", encoding="utf-8", buffering=1)
        # Dev-light keeps events.jsonl (filtered) but drops packets/inject_echo flood.
        drop_bulk = prod_mode or self.dev_light
        self._packets = (
            None
            if drop_bulk
            else self.packets_path.open("a", encoding="utf-8", buffering=1)
        )
        self._inject_echo = (
            None
            if drop_bulk
            else self.inject_echo_path.open("a", encoding="utf-8", buffering=1)
        )
        self._hw_debug = (
            None
            if drop_bulk
            else self.hw_debug_path.open("a", encoding="utf-8", buffering=1)
        )
        self.session_start_ns = time.perf_counter_ns()
        self.event_count = 0
        self.packet_count = 0
        self.glued_windows = 0
        self._packet_complete_hooks: list = []
        self._last_carrier: dict[str, int | None] = {}
        self._debug_log_rotate = debug_log_rotate_enabled(prod_mode=prod_mode)
        self._debug_log_max_bytes = debug_log_max_bytes()
        self._debug_log_utc_daily = debug_log_utc_daily_enabled()
        self._debug_log_utc_day = datetime.now(timezone.utc).strftime("%Y-%m-%d")
        self._debug_log_chunk = 0
        self._debug_log_rotating = False
        self._defer_debug_rotate = False
        # Recent eth→UART inject intervals for overlap tags on window/window_skip.
        self._recent_injects: list[dict[str, Any]] = []
        if self.hs_witness:
            print(
                f"[{self._witness_pos_tag()} witness] HS capture witness on — "
                "empty BSY (window_bytes=0) prints live",
                flush=True,
            )

    def _witness_pos_tag(self) -> str:
        pos = self.identity.fixture_pos
        return f"pos{pos}" if pos else "pos?"

    def _witness_track(self, event: str, fields: dict[str, Any]) -> None:
        if not self.hs_witness:
            return
        tag = self._witness_pos_tag()
        if event == "hs_inactive":
            self.witness.hs_inactive += 1
            wb = int(fields.get("window_bytes") or 0)
            hs_us = fields.get("hs_window_us")
            hs_us_s = f" hs_window_us={hs_us}" if hs_us is not None else ""
            if wb == 0:
                self.witness.hs_empty_bsy += 1
                print(
                    f"[{tag} witness] BSY empty #{self.witness.hs_empty_bsy} — "
                    f"hs_inactive window_bytes=0 "
                    f"({self.witness.hs_empty_bsy}/{self.witness.hs_inactive} empty) "
                    f"packets={self.packet_count} "
                    f"discarded={self.witness.bytes_discarded_inactive}"
                    f"{hs_us_s}",
                )
            else:
                self.witness.hs_with_data += 1
        elif event == "bytes_discarded_inactive":
            n = int(fields.get("count") or 0)
            self.witness.bytes_discarded_inactive += n
            print(
                f"[{tag} witness] bytes_discarded_inactive +{n} "
                f"(total={self.witness.bytes_discarded_inactive}) "
                f"first_byte={fields.get('first_byte', '?')}",
            )
        elif event == "hs_fifo_diag":
            self.witness.hs_fifo_diag += 1
            print(
                f"[{tag} witness] hs_fifo_diag "
                f"phase={fields.get('phase')} fifo_avail={fields.get('fifo_avail')} "
                f"drain_bytes={fields.get('drain_bytes')} cap_len={fields.get('cap_len')}",
            )
        elif event == "window_skip":
            self.witness.window_skip += 1
        elif event == "hs_uart_diag":
            return
        elif event == "packet_complete":
            self.witness.packet_complete += 1
            msg = str(fields.get("msg", "??"))
            self.witness.packet_complete_by_msg[msg] = (
                self.witness.packet_complete_by_msg.get(msg, 0) + 1
            )

    def print_witness_summary(self, *, force: bool = False) -> None:
        if not self.hs_witness:
            return
        w = self.witness
        if not force and w.hs_inactive == 0 and w.packet_complete == 0:
            return
        pc_parts = " ".join(
            f"0x{k}={v}" for k, v in sorted(w.packet_complete_by_msg.items())
        )
        pc_suffix = f" ({pc_parts})" if pc_parts else ""
        print(
            f"[{self._witness_pos_tag()} witness] summary: "
            f"hs={w.hs_inactive} empty_bsy={w.hs_empty_bsy} with_data={w.hs_with_data} "
            f"packets={self.packet_count} pc={w.packet_complete} "
            f"discarded={w.bytes_discarded_inactive} fifo_diag={w.hs_fifo_diag} "
            f"window_skip={w.window_skip}{pc_suffix}",
            flush=True,
        )

    def close(self) -> None:
        self._events.close()
        if self._packets is not None:
            self._packets.close()
        if self._inject_echo is not None:
            self._inject_echo.close()
        if getattr(self, "_hw_debug", None) is not None:
            self._hw_debug.close()

    def _reopen_capture_files(self) -> None:
        # Open-before-close: another thread may call emit() between close()
        # and the new open(), seeing a closed file handle if we close first.
        # By opening the new handle and swapping the attribute atomically
        # (single STORE_ATTR under the GIL), self._events is never closed
        # from any other thread's perspective.
        old = self._events
        self._events = self.events_path.open("a", encoding="utf-8", buffering=1)
        old.close()
        if self.prod_mode or self.dev_light:
            self._packets = None
            self._inject_echo = None
            self._hw_debug = None
            return
        if self._packets is not None:
            old_p = self._packets
            self._packets = self.packets_path.open("a", encoding="utf-8", buffering=1)
            old_p.close()
        if self._inject_echo is not None:
            old_ie = self._inject_echo
            self._inject_echo = self.inject_echo_path.open("a", encoding="utf-8", buffering=1)
            old_ie.close()
        if getattr(self, "_hw_debug", None) is not None:
            old_hw = self._hw_debug
            self._hw_debug = self.hw_debug_path.open("a", encoding="utf-8", buffering=1)
            old_hw.close()

    def set_defer_debug_rotate(self, defer: bool) -> None:
        """Skip rotate on inject/close hot path; call try_rotate_debug_logs when idle."""
        self._defer_debug_rotate = bool(defer)

    def try_rotate_debug_logs(self) -> None:
        """Idle/post-close rotate (safe when not under _tx_lock inject)."""
        if self.prod_mode or self._defer_debug_rotate:
            return
        if self._debug_log_rotate:
            self._maybe_rotate_debug_logs()

    def idle_flush(self) -> None:
        """Flush capture files when GPIO/HS is idle — not on the listen tick."""
        try:
            self._events.flush()
        except Exception:
            pass
        for handle in (self._packets, self._inject_echo, getattr(self, "_hw_debug", None)):
            if handle is None:
                continue
            try:
                handle.flush()
            except Exception:
                pass

    def _maybe_rotate_debug_logs(self) -> None:
        if self._defer_debug_rotate:
            return
        if not self._debug_log_rotate or self._debug_log_rotating:
            return
        self._debug_log_rotating = True
        try:
            meta = rotate_debug_capture_logs(
                self.output_dir,
                max_bytes=self._debug_log_max_bytes,
                utc_daily=self._debug_log_utc_daily,
                utc_day=self._debug_log_utc_day,
            )
            now_day = datetime.now(timezone.utc).strftime("%Y-%m-%d")
            if self._debug_log_utc_daily:
                self._debug_log_utc_day = now_day
            if meta is None:
                return
            self._debug_log_chunk += 1
            self._reopen_capture_files()
            row = {
                "utc": utc_iso(),
                **now_ns(),
                "fixture_pos": self.identity.fixture_pos,
                "pi_label": self.identity.pi_label,
                "role": self.identity.role,
                "ev": "debug_log_rotate",
                "chunk": self._debug_log_chunk,
                "max_bytes": self._debug_log_max_bytes,
                "utc_daily": self._debug_log_utc_daily,
                **meta,
            }
            self._events.write(json.dumps(row, separators=(",", ":")) + "\n")
            self._events.flush()
            self.event_count += 1
            if self.echo:
                print(json.dumps({"ev": "debug_log_rotate", **meta}), flush=True)
        finally:
            self._debug_log_rotating = False

    def _prod_should_log(self, event: str, fields: dict[str, Any]) -> tuple[bool, str]:
        if event == "status_0d_synth_skip":
            if fields.get("reason") not in PROD_SYNTH_SKIP_LOG_REASONS:
                return False, event
            return True, event
        if event in PROD_LOG_EVENTS:
            return True, event
        msg = str(fields.get("msg", "")).upper()
        if event == "packet_complete" and msg in PROD_CONFIG_TRANSFER_MSGS:
            return True, event
        if event in ("eth_rx_queued", "eth_tx", "eth_to_uart_done") and msg in PROD_CONFIG_TRANSFER_MSGS:
            return True, event
        if event == "inject_deferred" and msg in PROD_CONFIG_TRANSFER_MSGS:
            return True, event
        if event == "window_skip":
            try:
                window_bytes = int(fields.get("window_bytes", 0))
            except (TypeError, ValueError):
                window_bytes = 0
            if window_bytes in (32, 82, 1094, 1170) or window_bytes >= 500:
                return True, event
        if event in ("eth_rx_skip", "eth_tx_skip"):
            if (
                fields.get("reason") == "wire_len_crop"
                and msg in PROD_CONFIG_TRANSFER_MSGS
            ):
                return True, event
        if event == "link_carrier":
            port = str(fields.get("port", ""))
            carrier = fields.get("carrier")
            prev = self._last_carrier.get(port)
            self._last_carrier[port] = carrier
            if carrier == 0 and prev != 0:
                return True, "link_carrier_down"
            if carrier == 1 and prev == 0:
                return True, "link_carrier_up"
        return False, event

    @staticmethod
    def _dev_light_gameplay_wire_msg(msg: Any) -> bool:
        """True for wager / outcome / x1 ACK / status close — not 0A flood."""
        if msg is None:
            return False
        try:
            mt = int(str(msg), 16)
        except (TypeError, ValueError):
            return False
        return (
            is_gameplay_request(mt)
            or is_gameplay_response(mt)
            or is_gameplay_ack(mt)
            or mt == 0x0D
        )

    def _dev_light_should_log(self, event: str, fields: dict[str, Any]) -> tuple[bool, str]:
        """Faults + MRT/synth close path; suppress routine positive traffic."""
        if event == "status_0d_synth_skip":
            if fields.get("reason") not in DEV_LIGHT_SYNTH_SKIP_LOG_REASONS:
                return False, event
            return True, event
        if event == "inject_deferred":
            reason = str(fields.get("reason") or "")
            if reason in DEV_LIGHT_INJECT_DEFER_REASONS:
                return True, event
            return False, event
        if event == "transaction_defer":
            reason = str(fields.get("reason") or "")
            if reason in DEV_LIGHT_TRANSACTION_DEFER_REASONS:
                return True, event
            return False, event
        if event in DEV_LIGHT_LOG_EVENTS:
            return True, event
        # Gameplay UART↔ETH crumbs (Master 59 serve/retry, slave 51 uplink) — not 0A.
        if event in ("eth_tx", "eth_to_uart_done") and self._dev_light_gameplay_wire_msg(
            fields.get("msg")
        ):
            return True, event
        # Reuse prod window_skip / config-transfer / carrier filters.
        return self._prod_should_log(event, fields)

    def note_eth_inject(
        self,
        *,
        start_perf_ns: int,
        end_perf_ns: int,
        msg: str | None = None,
        tid: str | None = None,
        source: str = "eth_to_uart",
    ) -> None:
        """Record eth→UART inject time range for later window overlap tagging."""
        if self.prod_mode or self.dev_light:
            return
        self._recent_injects.append(
            {
                "start_perf_ns": int(start_perf_ns),
                "end_perf_ns": int(end_perf_ns),
                "msg": msg,
                "tid": tid,
                "source": source,
            }
        )
        # Keep a short ring — SCR forensics only need the last few seconds.
        if len(self._recent_injects) > 32:
            self._recent_injects = self._recent_injects[-32:]

    def inject_overlap_fields(
        self,
        window_start_perf_ns: Optional[int],
        window_end_perf_ns: Optional[int],
    ) -> dict[str, Any]:
        """Tag a capture window that overlaps a recent eth→UART inject."""
        if (
            window_start_perf_ns is None
            or window_end_perf_ns is None
            or not self._recent_injects
        ):
            return {}
        w0 = int(window_start_perf_ns)
        w1 = int(window_end_perf_ns)
        if w1 < w0:
            w0, w1 = w1, w0
        best: dict[str, Any] | None = None
        best_overlap = 0
        best_dt = 0.0
        for inj in self._recent_injects:
            i0 = int(inj["start_perf_ns"])
            i1 = int(inj["end_perf_ns"])
            # Pad inject end slightly — Master BSY/RCV INT may trail DE.
            i1_pad = i1 + 5_000_000
            ov0 = max(w0, i0)
            ov1 = min(w1, i1_pad)
            if ov1 <= ov0:
                continue
            ov = ov1 - ov0
            if ov > best_overlap:
                best_overlap = ov
                best = inj
                best_dt = (w0 - i0) / 1000.0
        if best is None:
            return {}
        out: dict[str, Any] = {
            "inject_overlap": True,
            "inject_overlap_msg": best.get("msg"),
            "inject_overlap_source": best.get("source"),
            "inject_overlap_us": round(best_overlap / 1000.0, 1),
            "inject_overlap_dt_us": round(best_dt, 1),
        }
        if best.get("tid"):
            out["inject_overlap_tid"] = best["tid"]
        return out

    def add_packet_complete_hook(self, hook) -> None:
        """hook(meta, pkt, packets_in_window, *, uart_first_byte: int | None) -> None"""
        self._packet_complete_hooks.append(hook)

    def emit(self, event: str, **fields: Any) -> None:
        log_event = event
        if self.prod_mode:
            ok, log_event = self._prod_should_log(event, fields)
            if not ok:
                return
            rotate_file_if_large(self.events_path, PROD_ERRORS_MAX_BYTES)
        elif self.dev_light:
            ok, log_event = self._dev_light_should_log(event, fields)
            if not ok:
                return
        row = {
            "utc": utc_iso(),
            **now_ns(),
            "fixture_pos": self.identity.fixture_pos,
            "pi_label": self.identity.pi_label,
            "role": self.identity.role,
            "ev": log_event,
            **fields,
        }
        line = json.dumps(row, separators=(",", ":")) + "\n"
        try:
            self._events.write(line)
        except ValueError:
            self._events.write(line)
        self.event_count += 1
        if not self.prod_mode:
            self._witness_track(event, fields)
        if self.echo:
            keys = (
                "ev",
                "msg",
                "bank",
                "local_bank",
                "bank_ok",
                "where",
                "message",
                "errno",
                "len",
                "tid",
                "dt_us",
                "peer_ip",
                "rtt_ms",
                "neigh",
                "port",
                "carrier",
                "eth_port",
                "ingress_port",
                "egress_port",
            )
            brief = {k: row[k] for k in keys if k in row}
            print(json.dumps(brief), flush=True)

    def write_packet_line(self, meta: PacketMeta) -> None:
        if self.prod_mode or self._packets is None:
            return
        self._packets.write(meta.hex_line + "\n")
        self.packet_count += 1

    def write_raw_window_line(self, raw: bytes) -> None:
        """Append one hex line = complete UART transmission (full HS window or inject)."""
        if self.prod_mode or self._packets is None or not raw:
            return
        self._packets.write(bytes_hex_line(raw) + "\n")
        self.packet_count += 1

    def write_inject_echo(
        self,
        sent: bytes,
        echo: bytes,
        *,
        source: str,
        engine: str,
        msg: str | None = None,
        tid: str | None = None,
        forensics: Any | None = None,
        copper: Any | None = None,
        post_de_rx: bytes | None = None,
        attempt: int | None = None,
        echo_verdict: str | None = None,
        full_hex: bool = False,
        emit_event: bool = True,
        wide_rx: bytes | None = None,
    ) -> None:
        """Log TX frame vs bus RX echo (+ uartd inject forensics when present).

        ``full_hex`` (dev soak): also put complete ``sent_hex`` / ``echo_hex`` on
        the events.jsonl ``inject_echo`` so archives can compare without opening
        inject_echo.jsonl. Bridge leaves this off (prefix only).
        """
        if self.prod_mode or self._inject_echo is None:
            return
        sent_hex = bytes_hex_line(sent)
        echo_hex = bytes_hex_line(echo) if echo else ""
        post = post_de_rx or b""
        post_hex = bytes_hex_line(post) if post else ""
        # Prefer inject-window echo; fall back to post-DE RX (XR often mute while DE on).
        bus_rx = echo if echo else post
        bus_rx_hex = bytes_hex_line(bus_rx) if bus_rx else ""
        echo_framing = raw_bus_framing_fields(echo, expect_len=len(sent) if sent else None)
        post_framing = raw_bus_framing_fields(post, expect_len=len(sent) if sent else None)
        bus_framing = raw_bus_framing_fields(bus_rx, expect_len=len(sent) if sent else None)
        match, match_kind = inject_echo_bytes_match(sent, echo)
        post_match, post_match_kind = inject_echo_bytes_match(sent, post)
        row: dict[str, Any] = {
            "utc": utc_iso(),
            **now_ns(),
            "fixture_pos": self.identity.fixture_pos,
            "pi_label": self.identity.pi_label,
            "role": self.identity.role,
            "ev": "inject_echo",
            "source": source,
            "engine": engine,
            "msg": msg,
            "tid": tid,
            "attempt": attempt,
            "sent_len": len(sent),
            "echo_len": len(echo),
            "echo_phase": "de_high",
            "match": match,
            "match_kind": match_kind,
            "sent": sent_hex,
            "echo": echo_hex,
            "post_de_rx_len": len(post),
            "post_de_rx": post_hex,
            "post_de_rx_match": post_match,
            "post_de_rx_match_kind": post_match_kind,
            "post_de_phase": "post_de",
            "bus_rx_raw": bus_rx_hex,
            "bus_rx_source": ("echo" if echo else ("post_de_rx" if post else "none")),
            "echo_soh_ok": echo_framing["soh_ok"],
            "echo_eop_ok": echo_framing["eop_ok"],
            "echo_framing_ok": echo_framing["framing_ok"],
            "post_de_soh_ok": post_framing["soh_ok"],
            "post_de_eop_ok": post_framing["eop_ok"],
            "post_de_framing_ok": post_framing["framing_ok"],
            "bus_soh_ok": bus_framing["soh_ok"],
            "bus_eop_ok": bus_framing["eop_ok"],
            "bus_framing_ok": bus_framing["framing_ok"],
            "bus_first_byte": bus_framing.get("first_byte", ""),
            "bus_last_byte": bus_framing.get("last_byte", ""),
            "bus_chop_hint": bus_framing.get("chop_hint"),
            **packet_bank_event_fields(wire_bank_number(sent)),
            **frame_marshal_fields(sent),
        }
        emit_extra: dict[str, Any] = {
            "marshal_crc32": row["marshal_crc32"],
            "marshal_len": row["marshal_len"],
            "post_de_rx_len": len(post),
            "post_de_rx_match": post_match,
            "post_de_rx_match_kind": post_match_kind,
            "match_kind": match_kind,
            "post_de_rx_prefix": bytes_hex_line(post[:16]) if post else "",
            "bus_rx_source": row["bus_rx_source"],
            "bus_soh_ok": row["bus_soh_ok"],
            "bus_eop_ok": row["bus_eop_ok"],
            "bus_framing_ok": row["bus_framing_ok"],
            "bus_first_byte": row["bus_first_byte"],
            "bus_last_byte": row["bus_last_byte"],
            "bus_chop_hint": row["bus_chop_hint"],
        }
        if wide_rx:
            from superball_slave_safety import decode_wide_rx_pairs, ninth_flag_hex

            wide_bytes = bytes(wide_rx)
            wide_data, ninth = decode_wide_rx_pairs(wide_bytes)
            wide_fields = wide_mark_framing_fields(wide_bytes)
            row.update(
                {
                    "echo_wide": bytes_hex_line(wide_bytes),
                    "echo_9th": ninth,
                    "echo_9th_n": len(ninth),
                    "echo_9th_flags": ninth_flag_hex(wide_bytes),
                    "echo_wide_data": bytes_hex_line(wide_data) if wide_data else "",
                    "echo_mark_only_first": bool(wide_fields.get("mark_only_first")),
                    "echo_mark_layout_ok": bool(wide_fields.get("mark_layout_ok")),
                    "echo_wide_is_wire": True,
                    "echo_mark_byte_indexes": wide_fields.get("mark_byte_indexes"),
                }
            )
            emit_extra.update(
                {
                    "echo_9th": ninth,
                    "echo_9th_n": len(ninth),
                    "echo_9th_flags": ninth_flag_hex(wide_bytes),
                    "echo_mark_only_first": row["echo_mark_only_first"],
                    "echo_mark_layout_ok": row["echo_mark_layout_ok"],
                    "echo_wide_is_wire": True,
                }
            )
        if attempt is not None:
            emit_extra["attempt"] = int(attempt)
        if forensics is not None:
            wide = getattr(forensics, "wide_bytes", b"") or b""
            tx_after_bulk = int(getattr(forensics, "tx_fifo_after_bulk", 0) or 0)
            tx_before = int(getattr(forensics, "tx_fifo_before", 0) or 0)
            tx_after_drain = int(getattr(forensics, "tx_fifo_after_drain", 0) or 0)
            tx_bulk_dec = decode_tx_fifo_sample(tx_after_bulk)
            row.update(
                {
                    "inject_method": int(getattr(forensics, "inject_method", 0) or 0),
                    "wide_is_wire": bool(getattr(forensics, "wide_is_wire", False)),
                    "driver_errno": int(getattr(forensics, "driver_errno", 0) or 0),
                    "forensics_frame_len": int(getattr(forensics, "frame_len", 0) or 0),
                    "wide_len": int(getattr(forensics, "wide_len", 0) or 0),
                    "rx_fifo_before": int(getattr(forensics, "rx_fifo_before", 0) or 0),
                    "rx_fifo_after": int(getattr(forensics, "rx_fifo_after", 0) or 0),
                    "drain_wait_us": round(
                        int(getattr(forensics, "drain_wait_ns", 0) or 0) / 1000.0, 1
                    ),
                    "tx_fifo_before": tx_before,
                    "tx_fifo_after_bulk": tx_after_bulk,
                    "tx_fifo_after_drain": tx_after_drain,
                    "tx_fifo_before_dec": decode_tx_fifo_sample(tx_before),
                    "tx_fifo_after_bulk_dec": tx_bulk_dec,
                    "tx_fifo_after_drain_dec": decode_tx_fifo_sample(tx_after_drain),
                    "usb_actual": int(getattr(forensics, "usb_actual", 0) or 0),
                    "de_hold_us": int(getattr(forensics, "de_hold_us", 0) or 0),
                    "tx_empty_reason": int(
                        getattr(forensics, "tx_empty_reason", None)
                        if getattr(forensics, "tx_empty_reason", None) is not None
                        else getattr(forensics, "lsr", 0) or 0
                    ),
                    "de_release_reason": int(getattr(forensics, "de_release_reason", 0) or 0),
                    "de_release_reason_name": str(
                        getattr(forensics, "de_release_reason_name", "")
                        or {
                            0: "none",
                            1: "temt",
                            2: "timeout",
                            3: "error",
                            4: "deadline",
                        }.get(int(getattr(forensics, "de_release_reason", 0) or 0), "unknown")
                    ),
                    "tx_empty_reason_name": {
                        0: "none",
                        1: "already_idle",
                        2: "en_act_fall",
                        3: "timeout_ok",
                        4: "spurious",
                    }.get(
                        int(
                            getattr(forensics, "tx_empty_reason", None)
                            if getattr(forensics, "tx_empty_reason", None) is not None
                            else getattr(forensics, "lsr", 0) or 0
                        ),
                        "unknown",
                    ),
                    "tx_fifo_not_empty_after_bulk": bool(
                        tx_bulk_dec.get("ok") and int(tx_bulk_dec.get("count") or 0) > 0
                    ),
                    "wide": bytes_hex_line(wide) if wide else "",
                    # wide_buf is rebuild of `sent` for logging — not a copper capture.
                    "wide_reconstructed": True,
                    **{
                        f"wide_{k}": v
                        for k, v in wide_mark_framing_fields(wide).items()
                        if k != "wide_len"
                    },
                }
            )
            emit_extra.update(
                {
                    "wide_is_wire": row["wide_is_wire"],
                    "wide_len": row["wide_len"],
                    "forensics_frame_len": row["forensics_frame_len"],
                    "rx_fifo_before": row["rx_fifo_before"],
                    "rx_fifo_after": row["rx_fifo_after"],
                    "driver_errno": row["driver_errno"],
                    "drain_wait_us": row["drain_wait_us"],
                    "tx_fifo_before": row["tx_fifo_before"],
                    "tx_fifo_after_bulk": row["tx_fifo_after_bulk"],
                    "tx_fifo_after_drain": row["tx_fifo_after_drain"],
                    "usb_actual": row["usb_actual"],
                    "de_hold_us": row.get("de_hold_us", 0),
                    "tx_empty_reason": row.get("tx_empty_reason", 0),
                    "tx_empty_reason_name": row.get("tx_empty_reason_name", ""),
                    "de_release_reason": row.get("de_release_reason_name", ""),
                    "tx_fifo_not_empty_after_bulk": row["tx_fifo_not_empty_after_bulk"],
                    "wide_mark_only_first": row.get("wide_mark_only_first"),
                    "wide_mark_layout_ok": row.get("wide_mark_layout_ok"),
                }
            )
        if copper is not None:
            samples = getattr(copper, "samples", None) or []
            copper_rows = [
                {
                    "phase": int(getattr(s, "phase", 0)),
                    "phase_name": getattr(s, "phase_name", str(getattr(s, "phase", ""))),
                    "t_us": int(getattr(s, "t_us", 0)),
                    "gpio_stat": f"0x{int(getattr(s, 'gpio_stat', 0)):04x}",
                    "gpio7": bool(getattr(s, "gpio7", False)),
                    "gpio1": bool(getattr(s, "gpio1", False)),
                    "tx_fifo": int(getattr(s, "tx_fifo", 0xffff)),
                    "rx_fifo": int(getattr(s, "rx_fifo", 0xffff)),
                }
                for s in samples
            ]
            row.update(
                {
                    "gpio_mode": f"0x{int(getattr(copper, 'gpio_mode', 0)):04x}",
                    "uart_flow": f"0x{int(getattr(copper, 'uart_flow', 0)):04x}",
                    "copper": copper_rows,
                    "copper_tx_fifo_peak": max(
                        (int(s["tx_fifo"]) for s in copper_rows if int(s["tx_fifo"]) != 0xFFFF),
                        default=None,
                    ),
                    "copper_gpio7_high": any(bool(s["gpio7"]) for s in copper_rows),
                    "copper_gpio1_high": any(bool(s["gpio1"]) for s in copper_rows),
                }
            )
            emit_extra.update(
                {
                    "gpio_mode": row["gpio_mode"],
                    "copper_n": len(copper_rows),
                    "copper_tx_fifo_peak": row["copper_tx_fifo_peak"],
                    "copper_gpio7_high": row["copper_gpio7_high"],
                    "copper_gpio1_high": row["copper_gpio1_high"],
                }
            )
        self._inject_echo.write(json.dumps(row, separators=(",", ":")) + "\n")
        if not emit_event:
            return
        if full_hex:
            last = f"{echo[-1]:02X}" if echo else None
            emit_extra.update(
                {
                    "sent_hex": sent_hex,
                    "echo_hex": echo_hex,
                    "echo_n": len(echo),
                    "sent_n": len(sent),
                    "echo_last": last,
                    "eop_ok": bool(echo_framing["eop_ok"]),
                    "echo_soh_ok": echo_framing["soh_ok"],
                    "echo_framing_ok": echo_framing["framing_ok"],
                    "len": len(sent),
                }
            )
            if echo_verdict is not None:
                emit_extra["echo"] = echo_verdict
        self.emit(
            "inject_echo",
            source=source,
            engine=engine,
            msg=msg,
            tid=tid,
            sent_len=len(sent),
            echo_len=len(echo),
            match=match,
            echo_prefix=bytes_hex_line(echo[:16]) if echo else "",
            **packet_bank_event_fields(wire_bank_number(sent)),
            **emit_extra,
        )

    def write_hw_debug(
        self,
        samples: list[dict[str, Any]],
        *,
        n: int,
        dropped: int,
    ) -> None:
        """Append compile-out HW debug timeline (from uartd IPC_HWDBG)."""
        if self.prod_mode or getattr(self, "_hw_debug", None) is None:
            return
        row = {
            "utc": utc_iso(),
            "fixture_pos": self.identity.fixture_pos,
            "pi_label": self.identity.pi_label,
            "role": self.identity.role,
            "ev": "hw_debug",
            "n": int(n),
            "dropped": int(dropped),
            "samples": samples,
        }
        self._hw_debug.write(json.dumps(row, separators=(",", ":")) + "\n")
        self.emit(
            "hw_debug",
            n=int(n),
            dropped=int(dropped),
            sample_types=[s.get("type_name") for s in samples[:16]],
        )

    def write_stats(self, extra: dict[str, Any]) -> None:
        if self.prod_mode:
            return
        payload = {
            "updated_at": utc_iso(),
            "session_start_perf_ns": self.session_start_ns,
            "fixture_pos": self.identity.fixture_pos,
            "pi_label": self.identity.pi_label,
            "role": self.identity.role,
            "event_count": self.event_count,
            "packet_count": self.packet_count,
            "glued_windows": self.glued_windows,
            "output_directory": str(self.output_dir),
            **({"hs_witness": self.witness.as_dict()} if self.hs_witness else {}),
            **extra,
        }
        self.stats_path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
        if self.hs_witness:
            self.print_witness_summary()

    def _emit_window_skip(
        self,
        buffer: bytearray,
        *,
        reason: str,
        normalized: bytes | None,
        window_start_perf_ns: Optional[int],
        first_byte_perf_ns: Optional[int],
        last_byte_perf_ns: Optional[int],
        uart_first_byte: int | None = None,
    ) -> None:
        raw = bytes(buffer)
        window_bytes = len(raw)
        # packets.raw already got the full copper window in finalize_handshake_window.
        fields: dict[str, Any] = {
            "window_bytes": window_bytes,
            "reason": reason,
            **window_skip_diag_fields(raw, normalized=normalized),
        }
        hex_prefix = bytes_hex_line(raw[:16]) if raw else ""
        if hex_prefix:
            fields["hex_prefix"] = hex_prefix
        if uart_first_byte is not None:
            fields["uart_first_byte"] = f"{uart_first_byte:02X}"
        if window_start_perf_ns is not None and last_byte_perf_ns is not None:
            fields["hs_window_us"] = round(
                (last_byte_perf_ns - window_start_perf_ns) / 1000.0, 1
            )
        if first_byte_perf_ns is not None and last_byte_perf_ns is not None:
            fields["byte_span_us"] = round(
                (last_byte_perf_ns - first_byte_perf_ns) / 1000.0, 1
            )
        fields.update(
            hs_porch_timing_fields(
                window_start_perf_ns=window_start_perf_ns,
                first_byte_perf_ns=first_byte_perf_ns,
                last_byte_perf_ns=last_byte_perf_ns,
                hs_inactive_perf_ns=None,
            )
        )
        fields.update(
            self.inject_overlap_fields(window_start_perf_ns, last_byte_perf_ns)
        )
        self.emit("window_skip", **fields)

    def finalize_handshake_window(
        self,
        buffer: bytearray,
        *,
        window_start_perf_ns: Optional[int],
        window_start_mono_ns: Optional[int],
        first_byte_perf_ns: Optional[int],
        last_byte_perf_ns: Optional[int],
        uart_first_byte: int | None = None,
        hs_inactive_perf_ns: Optional[int] = None,
    ) -> None:
        if not buffer:
            return
        window_bytes = len(buffer)
        raw = bytes(buffer)
        normalized = normalize_window(raw)
        if normalized is None:
            self.write_raw_window_line(raw)
            self._emit_window_skip(
                buffer,
                reason="no_05_0D",
                normalized=None,
                window_start_perf_ns=window_start_perf_ns,
                first_byte_perf_ns=first_byte_perf_ns,
                last_byte_perf_ns=last_byte_perf_ns,
                uart_first_byte=uart_first_byte,
            )
            return

        packets = split_packets(normalized)
        if not packets:
            self.write_raw_window_line(raw)
            self._emit_window_skip(
                buffer,
                reason="split_empty",
                normalized=normalized,
                window_start_perf_ns=window_start_perf_ns,
                first_byte_perf_ns=first_byte_perf_ns,
                last_byte_perf_ns=last_byte_perf_ns,
                uart_first_byte=uart_first_byte,
            )
            return

        porch = hs_porch_timing_fields(
            window_start_perf_ns=window_start_perf_ns,
            first_byte_perf_ns=first_byte_perf_ns,
            last_byte_perf_ns=last_byte_perf_ns,
            hs_inactive_perf_ns=hs_inactive_perf_ns,
        )
        dt_window_us: Optional[float] = None
        if window_start_perf_ns is not None and last_byte_perf_ns is not None:
            dt_window_us = (last_byte_perf_ns - window_start_perf_ns) / 1000.0
        if porch.get("hs_width_us") is not None:
            dt_window_us = float(porch["hs_width_us"])
        byte_span_us = porch.get("byte_span_us")
        local_bank = resolve_local_bank_number()
        parsed: list[tuple[int, bytes, Any, bool]] = []
        for idx, pkt in enumerate(packets):
            meta = parse_packet_meta(pkt, index_in_window=idx, count_in_window=len(packets))
            bank_ok = local_bank_accepts(meta.bank_number, local_bank=local_bank)
            parsed.append((idx, pkt, meta, bank_ok))
            if not bank_ok:
                continue
            for hook in self._packet_complete_hooks:
                hook(
                    meta,
                    pkt,
                    len(packets),
                    uart_first_byte=uart_first_byte,
                    window_us=round(dt_window_us, 1) if dt_window_us is not None else None,
                    byte_span_us=byte_span_us,
                )
        # Disk after hooks: leftover 170–214 B must not delay x9 → x1.
        self.write_raw_window_line(raw)
        if len(packets) > 1:
            self.glued_windows += 1
            self.emit(
                "window_glued",
                window_bytes=window_bytes,
                packet_count=len(packets),
                normalized_len=len(normalized),
            )
        if porch:
            self.emit(
                "hs_porch",
                window_bytes=window_bytes,
                packets_in_window=len(packets),
                **porch,
            )
        for idx, pkt, meta, bank_ok in parsed:
            framing = raw_bus_framing_fields(pkt)
            fields: dict[str, Any] = {
                "msg": meta.msg_type_hex,
                "len": len(pkt),
                "tid": meta.tid_hex,
                "game": meta.game_id_hex,
                "seq": meta.play_sequence,
                "counter": meta.session_counter,
                "window_bytes": window_bytes,
                "index_in_window": idx,
                "packets_in_window": len(packets),
                "soh_ok": framing["soh_ok"],
                "eop_ok": framing["eop_ok"],
                "first_byte": framing.get("first_byte", ""),
                "last_byte": framing.get("last_byte", ""),
                "hex_prefix": framing.get("hex_prefix", ""),
                "hex_suffix": framing.get("hex_suffix", ""),
                "checksum_ok": pog422_frame_checksum_ok(pkt),
            }
            fields.update(packet_bank_event_fields(meta, local_bank=local_bank))
            if meta.expected_response is not None:
                fields["expect_resp"] = f"{meta.expected_response:02X}"
            if meta.expected_ack is not None:
                fields["expect_ack"] = f"{meta.expected_ack:02X}"
            if meta.master_tid_hex is not None:
                fields["master_tid"] = meta.master_tid_hex
            if dt_window_us is not None:
                fields["window_us"] = round(dt_window_us, 1)
            fields.update(porch)
            fields.update(
                self.inject_overlap_fields(window_start_perf_ns, last_byte_perf_ns)
            )
            if not bank_ok:
                self.emit(
                    "packet_bank_drop",
                    msg=meta.msg_type_hex,
                    tid=meta.tid_hex,
                    len=len(pkt),
                    **packet_bank_event_fields(meta, local_bank=local_bank),
                )
            self.emit("packet_complete", **fields)

    def finalize_handshake_window_raw(
        self,
        buffer: bytearray,
        *,
        channel: BusChannel,
        window_start_perf_ns: Optional[int],
        first_byte_perf_ns: Optional[int],
        last_byte_perf_ns: Optional[int],
        uart_first_byte: int | None = None,
    ) -> None:
        """Log one GPIO HS window as raw hex — no SOH/EOP or COMMAND_WIRE_LEN assumptions."""
        if not buffer:
            return
        raw = bytes(buffer)
        window_bytes = len(raw)
        self.write_raw_window_line(raw)
        fields: dict[str, Any] = {
            "channel": channel.key,
            "len": window_bytes,
            "hex": bytes_hex_line(raw),
            "first_byte": f"{raw[0]:02X}",
            "last_byte": f"{raw[-1]:02X}",
        }
        if uart_first_byte is not None:
            fields["uart_first_byte"] = f"{uart_first_byte:02X}"
        if window_start_perf_ns is not None and last_byte_perf_ns is not None:
            fields["window_us"] = round(
                (last_byte_perf_ns - window_start_perf_ns) / 1000.0, 1
            )
        if first_byte_perf_ns is not None and last_byte_perf_ns is not None:
            fields["byte_span_us"] = round(
                (last_byte_perf_ns - first_byte_perf_ns) / 1000.0, 1
            )
        fields.update(
            self.inject_overlap_fields(window_start_perf_ns, last_byte_perf_ns)
        )
        self.emit("hs_window_raw", **fields)

    def finalize_handshake_window_for_channel(
        self,
        channel: BusChannel,
        buffer: bytearray,
        *,
        window_start_perf_ns: Optional[int],
        window_start_mono_ns: Optional[int],
        first_byte_perf_ns: Optional[int],
        last_byte_perf_ns: Optional[int],
        uart_first_byte: int | None = None,
        hs_inactive_perf_ns: Optional[int] = None,
    ) -> None:
        if channel_uses_raw_hs_window(channel):
            self.finalize_handshake_window_raw(
                buffer,
                channel=channel,
                window_start_perf_ns=window_start_perf_ns,
                first_byte_perf_ns=first_byte_perf_ns,
                last_byte_perf_ns=last_byte_perf_ns,
                uart_first_byte=uart_first_byte,
            )
        else:
            self.finalize_handshake_window(
                buffer,
                window_start_perf_ns=window_start_perf_ns,
                window_start_mono_ns=window_start_mono_ns,
                first_byte_perf_ns=first_byte_perf_ns,
                last_byte_perf_ns=last_byte_perf_ns,
                uart_first_byte=uart_first_byte,
                hs_inactive_perf_ns=hs_inactive_perf_ns,
            )


def bytes_hex_line(raw: bytes) -> str:
    return ",".join(f"{b:02X}" for b in raw)


def frame_marshal_fields(frame: bytes) -> dict[str, Any]:
    """Fingerprint ETH→UART payload at queue and at UART pull (integrity check)."""
    crc = zlib.crc32(frame) & 0xFFFFFFFF
    return {
        "marshal_crc32": f"{crc:08x}",
        "marshal_len": len(frame),
        "marshal_head": bytes_hex_line(frame[:8]) if frame else "",
        "marshal_tail": (
            bytes_hex_line(frame[-4:]) if len(frame) >= 4 else bytes_hex_line(frame)
        ),
    }


def decode_tx_fifo_sample(raw: int) -> dict[str, Any]:
    """Decode INJECT_TX TX_FIFO out-param (1W+ encodes get_reg failures)."""
    v = int(raw) & 0xFFFFFFFF
    out: dict[str, Any] = {"raw": v, "ok": False, "count": None}
    if v == 0xFFFFFFFF:
        out["fail"] = "opaque_ffffffff"
        return out
    if (v & 0xFF000000) == 0xE0000000:
        out["fail"] = "get_reg"
        out["uart_block"] = (v >> 16) & 0xFF
        out["errno"] = v & 0xFFFF
        return out
    out["ok"] = True
    out["count"] = v & 0x3FF
    return out


def raw_bus_framing_fields(raw: bytes, *, expect_len: int | None = None) -> dict[str, Any]:
    """Score raw UART bytes: SOH 0x05 head, EOP 0x0D tail, optional length."""
    if not raw:
        return {
            "raw_len": 0,
            "soh_ok": False,
            "eop_ok": False,
            "framing_ok": False,
            "hex_prefix": "",
            "hex_suffix": "",
        }
    soh_ok = raw[0] == 0x05
    eop_ok = raw[-1] == 0x0D
    mid_soh = sum(1 for b in raw[1:] if b == 0x05)
    mid_eop = sum(1 for b in raw[:-1] if b == 0x0D)
    out: dict[str, Any] = {
        "raw_len": len(raw),
        "first_byte": f"{raw[0]:02X}",
        "last_byte": f"{raw[-1]:02X}",
        "soh_ok": soh_ok,
        "eop_ok": eop_ok,
        "mid_soh_count": mid_soh,
        "mid_eop_count": mid_eop,
        "hex_prefix": bytes_hex_line(raw[:16]),
        "hex_suffix": bytes_hex_line(raw[-8:]) if len(raw) > 8 else bytes_hex_line(raw),
        "framing_ok": bool(soh_ok and eop_ok),
    }
    if expect_len is not None:
        out["len_ok"] = len(raw) == int(expect_len)
        out["framing_ok"] = bool(out["framing_ok"] and out["len_ok"])
        if len(raw) < int(expect_len):
            out["chop_hint"] = "short_missing_tail_or_head"
        elif len(raw) > int(expect_len):
            out["chop_hint"] = "long_extra_prefix_or_glued"
    return out


def wide_mark_framing_fields(wide: bytes) -> dict[str, Any]:
    """Wide USB buffer: MARK flag (0x01) must be only on the first data byte."""
    if len(wide) < 2 or (len(wide) % 2) != 0:
        return {"wide_len": len(wide), "mark_layout_ok": False, "mark_only_first": False}
    flags = [wide[i + 1] for i in range(0, len(wide), 2)]
    mark_idxs = [i for i, f in enumerate(flags) if (f & 0x01) != 0]
    only_first = mark_idxs == [0]
    return {
        "wide_len": len(wide),
        "mark_byte_indexes": mark_idxs,
        "mark_only_first": only_first,
        "mark_layout_ok": only_first and flags[0] == 0x01,
        "first_data": f"{wide[0]:02X}",
        "last_data": f"{wide[-2]:02X}",
    }


def window_skip_diag_fields(
    raw: bytes,
    *,
    normalized: bytes | None = None,
) -> dict[str, Any]:
    """Hex + opcode hint for unparsed GPIO HS windows (bench split_empty / no_05_0D)."""
    fields: dict[str, Any] = {}
    if not raw:
        return fields
    framing = raw_bus_framing_fields(raw)
    fields["hs_window_raw"] = bytes_hex_line(raw)
    fields["first_byte"] = framing.get("first_byte", f"{raw[0]:02X}")
    fields["last_byte"] = framing.get("last_byte", f"{raw[-1]:02X}")
    fields["soh_ok"] = framing["soh_ok"]
    fields["eop_ok"] = framing["eop_ok"]
    fields["hex_prefix"] = framing.get("hex_prefix", "")
    fields["hex_suffix"] = framing.get("hex_suffix", "")
    probe = normalized if normalized is not None else strip_before_mark(raw)
    if probe and len(probe) >= 2 and probe[0] == SOH:
        opcode = probe[1]
        fields["opcode_guess"] = f"{opcode:02X}"
        known = wire_lengths_for_cmd(opcode)
        if known:
            fields["wire_len_expected"] = list(known)
            fields["wire_len_short_by"] = [n - len(probe) for n in known]
    elif len(raw) >= 2:
        fields["opcode_guess"] = f"{raw[1]:02X}"
        fields["opcode_align"] = "raw_byte1"
    return fields


def channel_uses_raw_hs_window(channel: BusChannel) -> bool:
    """OIS: log full HS burst on inactive; framing/wakeup bit TBD."""
    if channel.key == "ois":
        return True
    return os.environ.get("FIXTURE_RAW_HS_WINDOW", "").strip() in ("1", "true", "yes")


@dataclass
class HandshakeCaptureState:
    handshake_active: bool = False
    buffer: bytearray = field(default_factory=bytearray)
    window_start_perf_ns: Optional[int] = None
    window_start_mono_ns: Optional[int] = None
    first_byte_perf_ns: Optional[int] = None
    last_byte_perf_ns: Optional[int] = None
    uart_first_byte: int | None = None
    bytes_collected: int = 0
    bytes_discarded_inactive: int = 0
    started_at: float = field(default_factory=time.time)
    post_x9_inject_grace_until_ns: Optional[int] = None
    post_x9_inject_grace_local_origin: bool = False
    post_x9_grace_soh_hold: bytearray = field(default_factory=bytearray)
    late_uart_hold: bytearray = field(default_factory=bytearray)
    capture_burst_on_fall: bool = False
    handoff_capture_x1_only: bool = False
    rx_snap_at_rise: dict[str, Any] | None = None
    window_max_in_waiting: int = 0
    window_max_fionread: int = 0
    # GPIO already fell; USB still owes COMMAND_WIRE_LEN for this opcode.
    short_frame_hold_until_ns: Optional[int] = None


def post_x9_grace_active(
    state: HandshakeCaptureState,
    now_ns: int | None = None,
) -> bool:
    """True while local post-x9 FIFO grace window is open (machine bridge state)."""
    until = state.post_x9_inject_grace_until_ns
    if until is None:
        return False
    if now_ns is None:
        now_ns = time.perf_counter_ns()
    return now_ns < until


def post_x9_local_grace_active(
    state: HandshakeCaptureState,
    now_ns: int | None = None,
) -> bool:
    """True during local-TID post-x9 grace (25 ms) — relay injects should defer."""
    return post_x9_grace_active(state, now_ns=now_ns) and state.post_x9_inject_grace_local_origin


def filter_handoff_capture_to_x1_only(buffer: bytearray) -> tuple[bytearray | None, str]:
    """Handoff burst: keep gameplay x1 only; reject progressive/status glue windows."""
    if not buffer:
        return None, "empty"
    normalized = normalize_window(bytes(buffer))
    if normalized is None:
        return None, "no_soh_eop"
    packets = split_packets(normalized)
    if not packets:
        return None, "split_empty"
    ack_packets = [
        p
        for p in packets
        if len(p) >= 2 and p[0] == SOH and is_gameplay_ack(p[1])
    ]
    if not ack_packets:
        return None, "no_x1"
    detail = "solo_x1" if len(ack_packets) == 1 else "first_x1"
    return bytearray(ack_packets[0]), detail


def _clear_post_x9_grace_state(state: HandshakeCaptureState) -> None:
    state.post_x9_inject_grace_until_ns = None
    state.post_x9_inject_grace_local_origin = False
    state.post_x9_grace_soh_hold.clear()


LATE_UART_HOLD_MAX = 8192
# USB leftover after GPIO fall; native 18+26 0x49 arrived 0.2–0.5 ms later.
# Must stay under native 41→0D min (~3.5 ms) so we do not glue the next phase.
SHORT_FRAME_HOLD_NS = 1_000_000


def _handshake_absorb_late_uart(
    ser: serial.Serial,
    channel: BusChannel,
    event_log: EventLog,
    state: HandshakeCaptureState,
    chunk: bytes,
    *,
    force: bool = False,
) -> None:
    """GPIO already inactive; UART is USB-late. Copper was sent with DE high."""
    buf = bytearray()
    if state.late_uart_hold:
        buf.extend(state.late_uart_hold)
        state.late_uart_hold.clear()
    if state.post_x9_grace_soh_hold:
        buf.extend(state.post_x9_grace_soh_hold)
        state.post_x9_grace_soh_hold.clear()
    if chunk:
        buf.extend(chunk)
    if len(buf) > LATE_UART_HOLD_MAX:
        buf = bytearray(buf[-LATE_UART_HOLD_MAX:])
    if not buf:
        return
    if EOP not in buf and not force:
        state.late_uart_hold.extend(buf)
        event_log.emit(
            "uart_after_hs_fall_hold",
            count=len(buf),
            first_byte=f"{buf[0]:02X}",
        )
        return
    _handshake_begin_window(channel, event_log, state, source="uart_after_hs_fall")
    now_ns = time.perf_counter_ns()
    state.first_byte_perf_ns = now_ns
    state.last_byte_perf_ns = now_ns
    state.uart_first_byte = buf[0]
    state.buffer.extend(buf)
    state.bytes_collected += len(buf)
    event_log.emit(
        "uart_first_byte",
        byte=f"{buf[0]:02X}",
        source="uart_after_hs_fall",
    )
    fields: dict[str, Any] = {
        "fifo_bytes": len(buf),
        "first_byte": f"{buf[0]:02X}",
        "forced": bool(force and EOP not in buf),
    }
    if len(buf) > 1:
        fields["second_byte"] = f"{buf[1]:02X}"
    event_log.emit("uart_after_hs_fall", **fields)
    if (
        post_x9_grace_active(state)
        and len(buf) >= 2
        and buf[0] == SOH
        and is_gameplay_ack(buf[1])
    ):
        event_log.emit(
            "post_x9_inject_grace_absorb",
            fifo_bytes=len(buf),
            msg=f"{buf[1]:02X}",
        )
    _handshake_finalize_window(ser, channel, event_log, state)


def _short_frame_hold_active(state: HandshakeCaptureState) -> bool:
    return state.short_frame_hold_until_ns is not None


def _emit_hs_inactive(
    channel: BusChannel,
    event_log: EventLog,
    state: HandshakeCaptureState,
) -> None:
    hs_fields: dict[str, Any] = {
        "gpio": channel.gpio_name,
        "silk": channel.legacy_silk,
        "window_bytes": len(state.buffer),
    }
    if state.window_start_perf_ns is not None:
        hs_fields["hs_window_us"] = round(
            (time.perf_counter_ns() - state.window_start_perf_ns) / 1000.0, 1
        )
    event_log.emit("hs_inactive", **hs_fields)


def _handshake_begin_window(
    channel: BusChannel,
    event_log: EventLog,
    state: HandshakeCaptureState,
    *,
    source: str | None = None,
) -> None:
    state.buffer.clear()
    state.window_start_perf_ns = time.perf_counter_ns()
    state.window_start_mono_ns = time.monotonic_ns()
    state.first_byte_perf_ns = None
    state.last_byte_perf_ns = None
    state.uart_first_byte = None
    state.rx_snap_at_rise = None
    state.window_max_in_waiting = 0
    state.window_max_fionread = 0
    state.short_frame_hold_until_ns = None
    fields: dict[str, Any] = {
        "gpio": channel.gpio_name,
        "silk": channel.legacy_silk,
    }
    if source:
        fields["source"] = source
    event_log.emit("hs_active", **fields)


def _handshake_commit_buffer(
    ser: serial.Serial,
    channel: BusChannel,
    event_log: EventLog,
    state: HandshakeCaptureState,
) -> None:
    """Parse RAM as a closed HS window. Does not emit hs_inactive."""
    del ser
    finalize_buf = state.buffer
    if state.handoff_capture_x1_only:
        state.handoff_capture_x1_only = False
        filtered, detail = filter_handoff_capture_to_x1_only(state.buffer)
        if filtered is None:
            event_log.emit(
                "post_x9_handoff_capture_reject",
                window_bytes=len(state.buffer),
                reason=detail,
                first_byte=f"{state.buffer[0]:02X}" if state.buffer else None,
            )
            state.bytes_discarded_inactive += len(state.buffer)
            state.buffer.clear()
            state.window_start_perf_ns = None
            state.window_start_mono_ns = None
            state.first_byte_perf_ns = None
            state.last_byte_perf_ns = None
            state.uart_first_byte = None
            state.handshake_active = False
            state.capture_burst_on_fall = False
            state.short_frame_hold_until_ns = None
            _clear_post_x9_grace_state(state)
            return
        if len(filtered) != len(state.buffer):
            event_log.emit(
                "post_x9_handoff_capture_filter",
                window_bytes=len(state.buffer),
                kept_bytes=len(filtered),
                reason=detail,
            )
        finalize_buf = filtered
    event_log.finalize_handshake_window_for_channel(
        channel,
        finalize_buf,
        window_start_perf_ns=state.window_start_perf_ns,
        window_start_mono_ns=state.window_start_mono_ns,
        first_byte_perf_ns=state.first_byte_perf_ns,
        last_byte_perf_ns=state.last_byte_perf_ns,
        uart_first_byte=state.uart_first_byte,
    )
    state.buffer.clear()
    state.window_start_perf_ns = None
    state.window_start_mono_ns = None
    state.first_byte_perf_ns = None
    state.last_byte_perf_ns = None
    state.uart_first_byte = None
    state.handshake_active = False
    state.capture_burst_on_fall = False
    state.rx_snap_at_rise = None
    state.window_max_in_waiting = 0
    state.window_max_fionread = 0
    state.short_frame_hold_until_ns = None
    _clear_post_x9_grace_state(state)


def _handshake_finalize_window(
    ser: serial.Serial,
    channel: BusChannel,
    event_log: EventLog,
    state: HandshakeCaptureState,
    *,
    drain_uart: bool = True,
    allow_short_hold: bool = True,
    emit_inactive: bool = True,
) -> None:
    chunk = uart_drain_no_usb(ser) if drain_uart else b""
    if chunk:
        state.buffer.extend(chunk)
        state.bytes_collected += len(chunk)
        state.last_byte_perf_ns = time.perf_counter_ns()
    window_bytes = len(state.buffer)
    # USB-late leftover is not Master still driving DE. Clear before any jsonl.
    state.handshake_active = False
    if (
        allow_short_hold
        and window_bytes
        and short_frame_bytes_needed(bytes(state.buffer)) is not None
    ):
        if emit_inactive:
            _emit_hs_inactive(channel, event_log, state)
        need = short_frame_bytes_needed(bytes(state.buffer))
        assert need is not None
        opcode, still = need
        if state.short_frame_hold_until_ns is None:
            state.short_frame_hold_until_ns = (
                time.perf_counter_ns() + SHORT_FRAME_HOLD_NS
            )
            event_log.emit(
                "short_frame_hold",
                opcode=f"{opcode:02X}",
                have=window_bytes,
                need=still,
            )
        return
    if emit_inactive:
        _emit_hs_inactive(channel, event_log, state)
        if window_bytes == 0 and drain_uart:
            event_log.emit(
                "hs_uart_diag",
                phase="empty_window",
                channel=channel.key,
                window_bytes=0,
                verdict="skipped_usb_snapshot",
            )
    _handshake_commit_buffer(ser, channel, event_log, state)


def _handshake_commit_short_hold(
    ser: serial.Serial,
    channel: BusChannel,
    event_log: EventLog,
    state: HandshakeCaptureState,
    *,
    reason: str,
) -> None:
    """End a GPIO-fall short hold: parse RAM, do not emit a second hs_inactive."""
    event_log.emit(
        "short_frame_hold_done",
        reason=reason,
        have=len(state.buffer),
    )
    _handshake_finalize_window(
        ser,
        channel,
        event_log,
        state,
        drain_uart=False,
        allow_short_hold=False,
        emit_inactive=False,
    )


def _handshake_short_hold_tick(
    ser: serial.Serial,
    channel: BusChannel,
    event_log: EventLog,
    state: HandshakeCaptureState,
) -> None:
    """HS already low. Append USB leftover until wire_len, timeout, or still short."""
    chunk = uart_drain_no_usb(ser)
    if chunk:
        state.buffer.extend(chunk)
        state.bytes_collected += len(chunk)
        state.last_byte_perf_ns = time.perf_counter_ns()
        event_log.emit(
            "short_frame_hold_append",
            n=len(chunk),
            have=len(state.buffer),
        )
    expired = time.perf_counter_ns() >= int(state.short_frame_hold_until_ns or 0)
    need = short_frame_bytes_needed(bytes(state.buffer))
    if need is None:
        _handshake_commit_short_hold(
            ser, channel, event_log, state, reason="complete"
        )
        return
    if expired:
        _handshake_commit_short_hold(
            ser, channel, event_log, state, reason="timeout"
        )


def arm_post_x9_inject_grace(
    event_log: EventLog,
    state: HandshakeCaptureState,
    *,
    slv_in_after_off: bool,
    local_origin_x9: bool = False,
) -> None:
    """Arm post-x9 FIFO grace after uartd INJECT_DONE (no pyserial fd)."""
    grace_ms = (
        POST_X9_INJECT_LOCAL_TID_FIFO_GRACE_MS
        if local_origin_x9
        else POST_X9_INJECT_FIFO_GRACE_MS
    )
    grace_ns = int(grace_ms * 1_000_000)
    if not slv_in_after_off and not local_origin_x9:
        _clear_post_x9_grace_state(state)
        return
    state.post_x9_inject_grace_until_ns = time.perf_counter_ns() + grace_ns
    if local_origin_x9:
        state.post_x9_inject_grace_local_origin = True
        if slv_in_after_off:
            state.handoff_capture_x1_only = True
    event_log.emit(
        "post_x9_inject_grace_arm",
        slv_in_live=bool(slv_in_after_off),
        local_origin=bool(local_origin_x9),
        engine="uartd",
    )


def handshake_post_x9_inject_handoff(
    ser: serial.Serial,
    channel: BusChannel,
    handshake: HandshakeMonitor,
    event_log: EventLog,
    state: HandshakeCaptureState,
    *,
    slv_in_after_off: bool,
    local_origin_x9: bool = False,
) -> None:
    """Mirror uartd cap_end_inject_window — drain echo when quiet; burst when SLV live."""
    live = slv_in_after_off or handshake.is_active()
    waiting = ser.in_waiting
    grace_ms = (
        POST_X9_INJECT_LOCAL_TID_FIFO_GRACE_MS
        if local_origin_x9
        else POST_X9_INJECT_FIFO_GRACE_MS
    )
    grace_ns = int(grace_ms * 1_000_000)

    if not live:
        if local_origin_x9:
            # Local wager round: terminal may assert x1 HS after GPIO poll — arm grace
            # without SLV-IN live (v16t did this for all x9; v16u limits to local origin).
            state.post_x9_inject_grace_until_ns = time.perf_counter_ns() + grace_ns
            state.post_x9_inject_grace_local_origin = True
            if waiting:
                drained = ser.read(waiting)
                state.bytes_discarded_inactive += len(drained)
                if drained:
                    event_log.emit(
                        "post_x9_inject_echo_drain",
                        count=len(drained),
                        first_byte=f"{drained[0]:02X}",
                        local_origin=True,
                    )
            event_log.emit(
                "post_x9_inject_grace_arm",
                slv_in_live=False,
                local_origin=True,
            )
            return
        _clear_post_x9_grace_state(state)
        if waiting:
            drained = ser.read(waiting)
            state.bytes_discarded_inactive += len(drained)
            if drained:
                event_log.emit(
                    "post_x9_inject_echo_drain",
                    count=len(drained),
                    first_byte=f"{drained[0]:02X}",
                )
        return
    state.post_x9_inject_grace_until_ns = time.perf_counter_ns() + grace_ns
    if local_origin_x9:
        state.post_x9_inject_grace_local_origin = True
        state.handoff_capture_x1_only = True

    if _short_frame_hold_active(state):
        _handshake_commit_short_hold(
            ser, channel, event_log, state, reason="gpio_rise"
        )
    if state.handshake_active and not state.capture_burst_on_fall:
        _handshake_finalize_window(ser, channel, event_log, state)

    _handshake_begin_window(
        channel, event_log, state, source="post_x9_inject_handoff"
    )
    state.handshake_active = True
    state.capture_burst_on_fall = True
    event_log.emit(
        "post_x9_inject_handoff",
        slv_in_live=True,
        slv_in_after_off=slv_in_after_off,
        fifo_bytes=0,
        burst_on_fall=True,
    )


def _drain_hs_edges(handshake: Any) -> list[dict[str, Any]]:
    fn = getattr(handshake, "drain_edges", None)
    if not callable(fn):
        return []
    try:
        return list(fn() or [])
    except OSError:
        return []


def _apply_hs_edges(
    ser: serial.Serial,
    channel: BusChannel,
    handshake: Any,
    event_log: EventLog,
    state: HandshakeCaptureState,
    edges: list[dict[str, Any]],
) -> None:
    """Apply kernel GPIO HS edges. Slave-fall then Master-rise must be two windows."""
    for i, edge in enumerate(edges):
        kind = edge.get("kind")
        later = edges[i + 1 :]
        if kind == "inactive" and state.handshake_active:
            followed_by_active = any(e.get("kind") == "active" for e in later)
            _handshake_finalize_window(
                ser,
                channel,
                event_log,
                state,
                drain_uart=not followed_by_active,
                allow_short_hold=not followed_by_active,
            )
        elif kind == "active":
            if _short_frame_hold_active(state):
                _handshake_commit_short_hold(
                    ser, channel, event_log, state, reason="gpio_rise"
                )
            if state.late_uart_hold or state.post_x9_grace_soh_hold:
                _handshake_absorb_late_uart(
                    ser, channel, event_log, state, b"", force=True
                )
            if state.handshake_active:
                _handshake_finalize_window(
                    ser, channel, event_log, state, drain_uart=False
                )
            _handshake_begin_window(channel, event_log, state, source="gpio_edge")
            state.handshake_active = True
            # No uart_rx_snapshot on the edge path — USB GET_REG hid 41→0D.
            state.rx_snap_at_rise = None
            state.window_max_in_waiting = 0
            state.window_max_fionread = 0


def handshake_capture_tick(
    ser: serial.Serial,
    channel: BusChannel,
    handshake: HandshakeMonitor,
    event_log: EventLog,
    state: HandshakeCaptureState,
) -> None:
    edges = _drain_hs_edges(handshake)
    if edges:
        _apply_hs_edges(ser, channel, handshake, event_log, state, edges)

    hs_active = handshake.is_active()
    now_ns = time.perf_counter_ns()

    until = state.post_x9_inject_grace_until_ns
    if until is not None and now_ns >= until:
        _clear_post_x9_grace_state(state)

    if _short_frame_hold_active(state) and not hs_active:
        _handshake_short_hold_tick(ser, channel, event_log, state)
        return

    if hs_active and not state.handshake_active and not state.capture_burst_on_fall:
        if _short_frame_hold_active(state):
            _handshake_commit_short_hold(
                ser, channel, event_log, state, reason="gpio_rise"
            )
        if state.late_uart_hold or state.post_x9_grace_soh_hold:
            _handshake_absorb_late_uart(
                ser, channel, event_log, state, b"", force=True
            )
        _handshake_begin_window(channel, event_log, state)
        state.handshake_active = True
        state.rx_snap_at_rise = None
        state.window_max_in_waiting = 0
        state.window_max_fionread = 0

    if state.handshake_active and not hs_active:
        _handshake_finalize_window(ser, channel, event_log, state)
        return

    if not state.capture_burst_on_fall:
        state.handshake_active = hs_active
    elif hs_active:
        state.handshake_active = True

    if not hs_active and not state.handshake_active:
        chunk = uart_drain_no_usb(ser)
        if chunk or state.late_uart_hold or state.post_x9_grace_soh_hold:
            _handshake_absorb_late_uart(ser, channel, event_log, state, chunk)
        return

    # Window ends on GPIO HS inactive, not UART idle. Never block on select()
    # and never uart_rx_snapshot in this hot path (USB GET_REG hid slave 0x41
    # then Master 0x0D as one window). Kernel GPIO edges are drained at tick
    # start; drain again after this UART read.
    if not handshake.is_active():
        if state.handshake_active:
            _handshake_finalize_window(ser, channel, event_log, state)
        return
    chunk = uart_drain_no_usb(ser)
    if not chunk:
        post = _drain_hs_edges(handshake)
        if post:
            _apply_hs_edges(ser, channel, handshake, event_log, state, post)
        return
    if state.first_byte_perf_ns is None:
        state.first_byte_perf_ns = time.perf_counter_ns()
        state.uart_first_byte = chunk[0]
        event_log.emit("uart_first_byte", byte=f"{chunk[0]:02X}")
    state.buffer.extend(chunk)
    state.bytes_collected += len(chunk)
    state.last_byte_perf_ns = time.perf_counter_ns()
    post = _drain_hs_edges(handshake)
    if post:
        _apply_hs_edges(ser, channel, handshake, event_log, state, post)
    elif not handshake.is_active() and state.handshake_active:
        _handshake_finalize_window(ser, channel, event_log, state)


# Cap rolling UART stream when ignore-hs listen cannot find SOH (mid-burst sync).
RAW_STREAM_MAX_BYTES = 16384

# Re-sync window when stream buffer grows without a decodable frame.
RAW_STREAM_RESYNC_KEEP = 4096


@dataclass
class RawCaptureState:
    stream: bytearray = field(default_factory=bytearray)
    bytes_collected: int = 0
    bytes_skipped_preamble: int = 0
    packets_emitted: int = 0
    uart_first_byte: int | None = None
    started_at: float = field(default_factory=time.time)


def drain_stream_packets(buf: bytearray) -> list[bytes]:
    """Pull complete length-aware packets from the front of a UART byte stream."""
    packets: list[bytes] = []
    while buf:
        while buf and buf[0] != SOH:
            del buf[0]
        if len(buf) < 2:
            break
        cmd = buf[1]
        known_lens = wire_lengths_for_cmd(cmd)
        framed = False
        for wire_len in known_lens:
            if len(buf) < wire_len:
                break
            pkt = bytes(buf[:wire_len])
            if pkt[0] == SOH and pkt[-1] == EOP:
                packets.append(pkt)
                del buf[:wire_len]
                framed = True
                break
        if framed:
            continue
        if known_lens:
            del buf[:1]
            continue
        eop_idx = buf.find(EOP, 2)
        if eop_idx < 0:
            break
        seg_len = eop_idx + 1
        packets.append(bytes(buf[:seg_len]))
        del buf[:seg_len]
    if len(buf) > RAW_STREAM_MAX_BYTES:
        mark = buf.rfind(bytes([SOH]))
        if mark > 0:
            del buf[:mark]
        elif len(buf) > RAW_STREAM_MAX_BYTES:
            del buf[: len(buf) - RAW_STREAM_RESYNC_KEEP]
    return packets


def _emit_stream_packets(
    event_log: EventLog,
    packets: list[bytes],
    state: RawCaptureState,
    *,
    source: str,
) -> None:
    for pkt in packets:
        if state.uart_first_byte is None:
            state.uart_first_byte = pkt[0]
            event_log.emit("uart_first_byte", byte=f"{pkt[0]:02X}", source=source)
        meta = parse_packet_meta(pkt, index_in_window=0, count_in_window=1)
        event_log.write_packet_line(meta)
        framing = raw_bus_framing_fields(pkt)
        fields: dict[str, Any] = {
            "msg": meta.msg_type_hex,
            "len": len(pkt),
            "tid": meta.tid_hex,
            "game": meta.game_id_hex,
            "seq": meta.play_sequence,
            "counter": meta.session_counter,
            "source": source,
            "soh_ok": framing["soh_ok"],
            "eop_ok": framing["eop_ok"],
            "first_byte": framing.get("first_byte", ""),
            "last_byte": framing.get("last_byte", ""),
            "hex_prefix": framing.get("hex_prefix", ""),
            "hex_suffix": framing.get("hex_suffix", ""),
        }
        fields.update(packet_bank_event_fields(meta))
        if meta.expected_response is not None:
            fields["expect_resp"] = f"{meta.expected_response:02X}"
        if meta.expected_ack is not None:
            fields["expect_ack"] = f"{meta.expected_ack:02X}"
        if meta.master_tid_hex is not None:
            fields["master_tid"] = meta.master_tid_hex
        if not local_bank_accepts(meta.bank_number):
            event_log.emit(
                "packet_bank_drop",
                msg=meta.msg_type_hex,
                tid=meta.tid_hex,
                len=len(pkt),
                source=source,
                **packet_bank_event_fields(meta),
            )
        event_log.emit("packet_complete", **fields)
        state.packets_emitted += 1


def raw_capture_tick(
    ser: serial.Serial,
    event_log: EventLog,
    state: RawCaptureState,
) -> None:
    """Listen-only: read UART continuously; ignore SLV-IN / GPIO handshake gating."""
    ready, _, _ = select.select([ser], [], [], 0.02)
    if not ready:
        return
    chunk = ser.read(ser.in_waiting or 1)
    if not chunk:
        return
    before_len = len(state.stream)
    state.stream.extend(chunk)
    state.bytes_collected += len(chunk)
    packets = drain_stream_packets(state.stream)
    consumed = before_len + len(chunk) - len(state.stream)
    payload = sum(len(p) for p in packets)
    state.bytes_skipped_preamble += consumed - payload
    if packets:
        _emit_stream_packets(event_log, packets, state, source="ignore_hs")


def get_mac(ifname: str) -> bytes:
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    info = fcntl.ioctl(sock.fileno(), 0x8927, struct.pack("256s", ifname[:15].encode()))
    return info[18:24]


def get_ifindex(ifname: str) -> int:
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    ifreq = struct.pack("16sH14s", ifname[:15].encode(), 0, b"")
    res = fcntl.ioctl(sock.fileno(), 0x8933, ifreq)
    return struct.unpack("=I", res[16:20])[0]


def resolve_eth_port_names(*, eth_ports: str | None, legacy_interface: str | None) -> list[str]:
    """Pick AF_PACKET interfaces for directional routing."""
    if legacy_interface:
        names = [legacy_interface]
    elif eth_ports:
        names = [part.strip() for part in eth_ports.split(",") if part.strip()]
    else:
        names = ["eth0", "eth1"]
    live = [name for name in names if os.path.isdir(f"/sys/class/net/{name}")]
    if live:
        return live
    raise SystemExit(f"No network interfaces found from {names!r}")


# linux/if_packet.h — sll_pkttype on AF_PACKET recvmsg sockaddr_ll
PACKET_HOST = 0
PACKET_BROADCAST = 1
PACKET_MULTICAST = 2
PACKET_OTHERHOST = 3
PACKET_OUTGOING = 4
PACKET_LOOPBACK = 5

_PKTTYPE_DROP = frozenset({PACKET_OUTGOING, PACKET_LOOPBACK})


def sockaddr_ll_pkttype(addr: object) -> int | None:
    if not isinstance(addr, bytes) or len(addr) < 11:
        return None
    return addr[10]


def sockaddr_ll_ifindex(addr: object) -> int | None:
    if not isinstance(addr, bytes) or len(addr) < 8:
        return None
    return struct.unpack_from("=I", addr, 4)[0]


def eth_l2_src_mac(packet: bytes) -> bytes | None:
    if len(packet) < 12:
        return None
    return packet[6:12]


def recv_eth_l2_frame(
    sock: socket.socket,
) -> tuple[bytes | None, int | None, int | None, str | None]:
    """Receive one L2 frame; drop OUTGOING/LOOPBACK pkttype."""
    try:
        data, _anc, _flags, addr = sock.recvmsg(4096, 256)
    except BlockingIOError:
        return None, None, None, None
    except OSError:
        return None, None, None, None
    pkttype = sockaddr_ll_pkttype(addr)
    ifindex = sockaddr_ll_ifindex(addr)
    if pkttype in _PKTTYPE_DROP:
        return None, pkttype, ifindex, "outgoing_pkttype"
    return data, pkttype, ifindex, None


@dataclass
class EthPort:
    name: str
    sock: socket.socket
    ifindex: int
    mac: bytes
    header: bytes


class DirectionalEthRouter:
    """Per-interface raw sockets: UART → live ETH; ETH ingress → peer ETH + optional UART."""

    _send_error_log_at: dict[str, float] = {}

    def __init__(
        self,
        ports: list[EthPort],
        *,
        uart_live_ports_only: bool = True,
        eth_relay_live_ports_only: bool = True,
        uart_egress_port: str | None = None,
    ) -> None:
        if not ports:
            raise ValueError("DirectionalEthRouter requires at least one ETH port")
        self.ports = {port.name: port for port in ports}
        self.ifindex_to_name = {port.ifindex: port.name for port in ports}
        self.sock_to_name = {port.sock: port.name for port in ports}
        self._local_macs = frozenset(port.mac for port in ports)
        self.uart_live_ports_only = uart_live_ports_only
        self.eth_relay_live_ports_only = eth_relay_live_ports_only
        if uart_egress_port is not None and uart_egress_port not in self.ports:
            raise ValueError(f"uart_egress_port {uart_egress_port!r} not in {list(self.ports)!r}")
        self.uart_egress_port = uart_egress_port
        # EG 2.2.2.71: TID → egress port for hub TID-unicast (learned from eth_rx).
        self.tid_to_port: dict[str, str] = {}
        self._fanout_log_min_ports = ETH_FANOUT_LOG_MIN_PORTS

    @classmethod
    def open(
        cls,
        port_names: list[str],
        *,
        uart_live_ports_only: bool = True,
        eth_relay_live_ports_only: bool = True,
        uart_egress_port: str | None = None,
    ) -> DirectionalEthRouter:
        ports: list[EthPort] = []
        for name in port_names:
            if not os.path.isdir(f"/sys/class/net/{name}"):
                continue
            sock = create_eth_socket(name)
            ifindex = get_ifindex(name)
            mac = get_mac(name)
            header = build_eth_header_from_mac(mac)
            ports.append(EthPort(name=name, sock=sock, ifindex=ifindex, mac=mac, header=header))
        if not ports:
            raise SystemExit(f"No ETH ports available from {port_names!r}")
        return cls(
            ports,
            uart_live_ports_only=uart_live_ports_only,
            eth_relay_live_ports_only=eth_relay_live_ports_only,
            uart_egress_port=uart_egress_port,
        )

    def port_names(self) -> list[str]:
        return list(self.ports.keys())

    def sockets(self) -> list[socket.socket]:
        # Skip closed/invalid fds — hub enroll reopen_port can close a sock while
        # eth_to_uart is between sockets() and select (ValueError: fd -1).
        out: list[socket.socket] = []
        for port in self.ports.values():
            try:
                if port.sock.fileno() >= 0:
                    out.append(port.sock)
            except OSError:
                continue
        return out

    def port_for_ifindex(self, ifindex: int | None) -> str | None:
        if ifindex is None:
            return None
        return self.ifindex_to_name.get(ifindex)

    def port_for_socket(self, sock: socket.socket) -> str | None:
        return self.sock_to_name.get(sock)

    def resolve_ingress_port(
        self,
        sock: socket.socket,
        ifindex: int | None,
    ) -> str | None:
        """Map AF_PACKET recv to interface name (ifindex often missing on Pi recvmsg)."""
        return self.port_for_ifindex(ifindex) or self.port_for_socket(sock)

    def is_local_loopback(self, packet: bytes) -> bool:
        src = eth_l2_src_mac(packet)
        return src is not None and src in self._local_macs

    @staticmethod
    def port_has_carrier(name: str) -> bool:
        try:
            return int(Path(f"/sys/class/net/{name}/carrier").read_text().strip()) == 1
        except OSError:
            return True

    def note_tid_port(self, tid_hex: str | None, port_name: str | None) -> None:
        """Learn TID→port from eth ingress (hub spoke mapping)."""
        if not tid_hex or not port_name:
            return
        key = tid_hex.upper()
        if key == BANK_PROGRESSIVE_BROADCAST_TID:
            return
        if port_name not in self.ports:
            return
        self.tid_to_port[key] = port_name

    def port_for_tid(self, tid_hex: str | None) -> str | None:
        if not tid_hex:
            return None
        return self.tid_to_port.get(tid_hex.upper())

    def send_uart_to_eth(
        self,
        payload: bytes,
        *,
        only_ports: list[str] | None = None,
    ) -> list[tuple[str, float]]:
        if self.uart_egress_port is not None:
            port = self.ports[self.uart_egress_port]
            t0 = time.perf_counter_ns()
            port.sock.send(port.header + payload)
            dt_us = (time.perf_counter_ns() - t0) / 1000.0
            return [(port.name, dt_us)]
        results, _stats = self.send_eth_to_peers(
            payload,
            live_ports_only=self.uart_live_ports_only,
            only_ports=only_ports,
        )
        return results

    def send_eth_to_peers(
        self,
        payload: bytes,
        *,
        except_port: str | None = None,
        live_ports_only: bool = False,
        only_ports: list[str] | None = None,
    ) -> tuple[list[tuple[str, float]], dict[str, Any]]:
        """Send to peers. Returns (per-port results, fanout_stats).

        fanout_stats: port_count, wall_us (first→last send), mode (flood|unicast).
        """
        results: list[tuple[str, float]] = []
        t_wall0 = time.perf_counter_ns()
        # None = flood all; [] = send nowhere (do not treat empty as flood).
        only_set = set(only_ports) if only_ports is not None else None
        for port in self.ports.values():
            if except_port is not None and port.name == except_port:
                continue
            if only_set is not None and port.name not in only_set:
                continue
            if live_ports_only and not self.port_has_carrier(port.name):
                continue
            t0 = time.perf_counter_ns()
            try:
                port.sock.send(port.header + payload)
            except OSError as exc:
                now = time.monotonic()
                last = self._send_error_log_at.get(port.name, 0.0)
                if now - last >= 30.0:
                    self._send_error_log_at[port.name] = now
                    print(
                        f"eth_send_error port={port.name} err={exc}",
                        file=sys.stderr,
                        flush=True,
                    )
                continue
            dt_us = (time.perf_counter_ns() - t0) / 1000.0
            results.append((port.name, dt_us))
        wall_us = (time.perf_counter_ns() - t_wall0) / 1000.0
        stats = {
            "port_count": len(results),
            "wall_us": wall_us,
            "mode": "unicast" if only_set is not None and len(only_set) <= 1 else "flood",
        }
        return results, stats

    def reopen_port(self, name: str) -> bool:
        """Close and recreate AF_PACKET socket for one interface (hot-plug recovery)."""
        if name not in self.ports:
            return False
        old = self.ports[name]
        try:
            old.sock.close()
        except OSError:
            pass
        self.sock_to_name.pop(old.sock, None)
        if not os.path.isdir(f"/sys/class/net/{name}"):
            return False
        try:
            sock = create_eth_socket(name)
            ifindex = get_ifindex(name)
            mac = get_mac(name)
            header = build_eth_header_from_mac(mac)
        except (OSError, SystemExit):
            return False
        port = EthPort(name=name, sock=sock, ifindex=ifindex, mac=mac, header=header)
        self.ports[name] = port
        self.ifindex_to_name[ifindex] = name
        self.sock_to_name[sock] = name
        self._local_macs = frozenset(p.mac for p in self.ports.values())
        return True

    def add_port(self, name: str) -> bool:
        """Hot-add one interface into a live router (hub enroll of eg-sNN).

        Idempotent: existing name → reopen. Used so Master can learn field spokes
        without restarting the bridge.
        """
        if not name:
            return False
        if name in self.ports:
            return self.reopen_port(name)
        if not os.path.isdir(f"/sys/class/net/{name}"):
            return False
        try:
            sock = create_eth_socket(name)
            ifindex = get_ifindex(name)
            mac = get_mac(name)
            header = build_eth_header_from_mac(mac)
        except (OSError, SystemExit):
            return False
        port = EthPort(name=name, sock=sock, ifindex=ifindex, mac=mac, header=header)
        self.ports[name] = port
        self.ifindex_to_name[ifindex] = name
        self.sock_to_name[sock] = name
        self._local_macs = frozenset(p.mac for p in self.ports.values())
        return True

    def remove_port(self, name: str) -> bool:
        """Hot-drop one interface (dead / swapped spoke). Idempotent if missing."""
        if not name or name not in self.ports:
            return False
        port = self.ports.pop(name)
        self.ifindex_to_name.pop(port.ifindex, None)
        self.sock_to_name.pop(port.sock, None)
        try:
            port.sock.close()
        except OSError:
            pass
        self.tid_to_port = {tid: p for tid, p in self.tid_to_port.items() if p != name}
        self._local_macs = frozenset(p.mac for p in self.ports.values())
        if self.uart_egress_port == name:
            self.uart_egress_port = None
        return True

    def close(self) -> None:
        for port in self.ports.values():
            try:
                port.sock.close()
            except Exception:
                pass


def create_eth_socket(interface: str) -> socket.socket:
    try:
        sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL))
    except PermissionError as exc:
        raise SystemExit(
            "AF_PACKET raw socket denied (need cap_net_raw on python3). "
            "Run: sudo bash ../prep_eth_for_fixture.sh"
        ) from exc
    sock.bind((interface, 0))
    return sock


def trim_to_last_eop(packet: bytes) -> Optional[bytes]:
    if not packet:
        return None
    frames = split_packets(packet)
    if frames:
        return frames[-1]
    try:
        end = packet.rindex(EOP)
        return packet[: end + 1]
    except ValueError:
        return None


def build_eth_header(interface: str) -> bytes:
    return build_eth_header_from_mac(get_mac(interface))


def build_eth_header_from_mac(src_mac: bytes) -> bytes:
    dest_mac = b"\xff\xff\xff\xff\xff\xff"
    eth_type = struct.pack("!H", ETH_TYPE)
    return dest_mac + src_mac + eth_type



def fixture_pos_arg(value: str) -> int:
    """CLI/env FIXTURE_POS: 1–99 (witness ids like 51 allowed). Log/socket routing only."""
    try:
        pos = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("FIXTURE_POS must be an integer 1..99") from exc
    if not 1 <= pos <= 99:
        raise argparse.ArgumentTypeError("FIXTURE_POS must be 1..99")
    return pos


def resolve_fixture_pos(cli: int | None = None) -> int:
    """Optional CLI/env log-routing id; never a run gate. Default 1 when unset/invalid."""
    if cli is not None:
        if isinstance(cli, int) and 1 <= cli <= 99:
            return cli
        return fixture_pos_arg(str(cli))
    raw = (os.environ.get("FIXTURE_POS") or "").strip()
    if raw:
        try:
            return fixture_pos_arg(raw)
        except argparse.ArgumentTypeError:
            pass
    return 1


def add_common_cli(parser: argparse.ArgumentParser) -> None:
    parser.add_argument(
        "--channel",
        choices=sorted(BUS_CHANNELS),
        default="machine",
    )
    parser.add_argument("--port", default=None)
    parser.add_argument("--output-dir", default=DATA_DIR)
    parser.add_argument("--logs-dir", default=LOGS_DIR)
    parser.add_argument(
        "--fixture-pos",
        type=fixture_pos_arg,
        default=None,
        metavar="N",
        help=(
            "Optional log/socket routing id 1–99 (Dropbox logs/N, uartd sock). "
            "Not a run gate. Env FIXTURE_POS or default 1 when omitted."
        ),
    )
    parser.add_argument("--pi-label", default="", help="Optional hostname/label in events")
    parser.add_argument("--active-high", action="store_true")
    parser.add_argument("--echo-events", action="store_true", help="Print compact events to stdout")
    parser.add_argument(
        "--no-hs-witness",
        action="store_true",
        help="Disable live HS/BSY console witness (empty window_bytes=0 counter)",
    )
    parser.add_argument("--stats-interval", type=float, default=30.0)
    parser.add_argument(
        "--append-capture",
        action="store_true",
        help="Keep existing data/logs files (default: clear on each start)",
    )
    parser.add_argument(
        "--prod",
        action="store_true",
        help="Production long-run: errors.jsonl only, no packets.raw, no stats",
    )
    parser.add_argument(
        "--dev-light",
        action="store_true",
        help="Fault-only events.jsonl (MRT/synth/errors); no packets.raw. "
        "Also FIXTURE_DEV_LIGHT=1 or FIXTURE_LOG_MODE=dev_light",
    )