ADN-DMR-Peer-Server/tests/routing/test_obp_concurrent_streams_downlink.py

# ADN DMR Peer Server - concurrent OBP streams downlink reproduction
#
# Copyright (C) 2026  Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>

"""Reproduce production bug: two concurrent OBP streams on one MASTER TS2 with a
dual-TG hotspot (SINGLE=0) intermittently drops the active listen stream.

Diagnosis-only: no production code changes. Uses DeterministicScenario for OBP
ingress + real HBPProtocol send_peer downlink gate.
"""

from __future__ import annotations

import copy
import logging
from contextlib import contextmanager
from typing import Any

import pytest
from tests.harness.deterministic import (
    DeterministicScenario,
    FakeClock,
    PacketSpec,
    add_openbridge_system,
    patch_routing_wall_time,
)
from tests.support.hbp_repeat_stack import RecordingTransport

from adn_server.application.routing.downlink import (
    DownlinkContext,
    peer_slot_blocks_downlink,
)
from adn_server.application.routing.helpers import (
    _peer_status_rx_hangtime_blocks,
    _peer_transmit_hangtime_blocks,
    hbp_slot_blocks_group_voice_for_peer,
    peer_hotspot_voice_slot_busy,
    peer_single_blocks_foreign_same_tg_downlink,
    peer_single_same_tg_foreign_tx_blocks,
    slot_has_active_voice,
)
from adn_server.domain import HBPF_DATA_SYNC, HBPF_SLT_VHEAD, HBPF_SLT_VTERM, bytes_3, bytes_4, int_id
from adn_server.domain.hbp_protocol import HBPF_VOICE, STREAM_TO
from adn_server.infrastructure.acl_router import InMemoryAclRouter
from adn_server.infrastructure.config_normalizer import ensure_system_runtime_config
from adn_server.infrastructure.twisted_adapters.udp_hbp import HBPProtocol

_TG_A = 3109050
_TG_B = 52090
_STREAM_A = 0xAABBCCDD
_STREAM_B = 0x11223344
_HS_PEER = 730039101
_OBP_PEER = 73010
_RF_A = 3340001
_RF_B = 3340002
_INTERVAL_S = 0.06
_VOICE_BURSTS = 8


def _bridge_row(*, system: str, ts: int, tgid: int) -> dict[str, Any]:
    tg_b = bytes_3(tgid)
    return {
        "SYSTEM": system,
        "TS": ts,
        "TGID": tg_b,
        "ACTIVE": True,
        "TIMEOUT": 3600.0,
        "TO_TYPE": "ON",
        "ON": [tg_b],
        "OFF": [],
        "RESET": [],
        "TIMER": 0.0,
    }


def _dual_tg_routing_table() -> dict[str, list[dict[str, Any]]]:
    return {
        str(_TG_A): [
            _bridge_row(system="OBP-CL", ts=1, tgid=_TG_A),
            _bridge_row(system="MASTER-A", ts=2, tgid=_TG_A),
        ],
        str(_TG_B): [
            _bridge_row(system="OBP-CL", ts=1, tgid=_TG_B),
            _bridge_row(system="MASTER-A", ts=2, tgid=_TG_B),
        ],
    }


@contextmanager
def _patch_harness_time(clock: FakeClock):
    """Patch wall clock on routing ingress and downlink gate paths."""
    import adn_server.application.routing.downlink as downlink_mod
    import adn_server.application.routing_use_cases as routing_mod

    orig_routing = routing_mod.time.time
    orig_downlink = downlink_mod.time.time
    routing_mod.time.time = clock.time
    downlink_mod.time.time = clock.time
    try:
        yield
    finally:
        routing_mod.time.time = orig_routing
        downlink_mod.time.time = orig_downlink


def _build_stack() -> tuple[DeterministicScenario, HBPProtocol, RecordingTransport]:
    routing_table = _dual_tg_routing_table()
    scenario = DeterministicScenario(
        routing_table=routing_table,
        enable_reporting=True,
    )
    config = scenario.config
    add_openbridge_system(config, "OBP-CL")
    master = config["SYSTEMS"]["MASTER-A"]
    master.update(
        {
            "MAX_PEERS": 8,
            "GROUP_HANGTIME": 5.0,
            "SINGLE_MODE": False,
            "TS2_STATIC": f"{_TG_A},{_TG_B}",
        }
    )
    ensure_system_runtime_config(config)

    transport = RecordingTransport()
    hbp = HBPProtocol("MASTER-A", config, router=InMemoryAclRouter())
    hbp.transport = transport  # type: ignore[assignment]
    scenario.protocols["MASTER-A"] = hbp

    def _send_to_system(target: str, packet: bytes, **kwargs: Any) -> None:
        scenario.capture.recorder(target)(
            packet,
            hops=kwargs.get("_hops", kwargs.get("hops", b"")),
            ber=kwargs.get("_ber", kwargs.get("ber", b"\x00")),
            rssi=kwargs.get("_rssi", kwargs.get("rssi", b"\x00")),
            source_server=kwargs.get(
                "_source_server", kwargs.get("source_server", b"\x00\x00\x00\x00"),
            ),
            source_rptr=kwargs.get(
                "_source_rptr", kwargs.get("source_rptr", b"\x00\x00\x00\x00"),
            ),
        )
        proto = scenario.protocols.get(target)
        if proto is not None and hasattr(proto, "send_system"):
            proto.send_system(
                packet,
                _hops=kwargs.get("_hops", b""),
                _ber=kwargs.get("_ber", b"\x00"),
                _rssi=kwargs.get("_rssi", b"\x00"),
                _source_server=kwargs.get("_source_server", b"\x00\x00\x00\x00"),
                _source_rptr=kwargs.get("_source_rptr", b"\x00\x00\x00\x00"),
            )

    scenario.routing._send_to_system = _send_to_system

    hs = bytes_4(_HS_PEER)
    hbp._peers[hs] = {
        "CONNECTION": "YES",
        "CONNECTED": scenario.clock.time(),
        "LAST_PING": scenario.clock.time(),
        "SOCKADDR": ("127.0.0.1", 62031),
        "OPTIONS": f"TS2={_TG_A},{_TG_B};SINGLE=0;".encode(),
    }
    master.setdefault("PEERS", {})[hs] = hbp._peers[hs]
    hbp._refresh_connected_peer_count()
    hbp._mark_downlink_index_dirty()
    return scenario, hbp, transport


def _base_spec(*, tgid: int, stream_id: int, rf_src: int) -> PacketSpec:
    return PacketSpec(
        peer_id=_OBP_PEER,
        rf_src=rf_src,
        dst_id=tgid,
        slot=1,
        stream_id=stream_id,
    )


def _dmrd_packet(
    *,
    tgid: int,
    stream_id: int,
    rf_src: int,
    frame_type: int,
    dtype_vseq: int,
    seq: int = 0,
) -> bytes:
    base = _base_spec(tgid=tgid, stream_id=stream_id, rf_src=rf_src)
    spec = PacketSpec(
        peer_id=base.peer_id,
        rf_src=base.rf_src,
        dst_id=base.dst_id,
        slot=base.slot,
        stream_id=base.stream_id,
        seq=seq,
        frame_type=frame_type,
        dtype_vseq=dtype_vseq,
        payload=base.payload,
    )
    return spec.data()


def _is_voice_burst(packet: bytes) -> bool:
    return ((packet[15] & 0x30) >> 4) == HBPF_VOICE


def _parse_dmrd_args(packet: bytes) -> dict[str, Any]:
    bits = packet[15]
    return {
        "peer_id": packet[11:15],
        "rf_src": packet[5:8],
        "dst_id": packet[8:11],
        "seq": packet[4],
        "slot": 2 if bits & 0x80 else 1,
        "call_type": "group",
        "frame_type": (bits & 0x30) >> 4,
        "dtype_vseq": bits & 0xF,
        "stream_id": packet[16:20],
        "data": packet,
    }


def _wrap_send_peer_trace(hbp: HBPProtocol) -> dict[str, Any]:
    """Record per-packet send_peer accept/reject without modifying production code."""
    trace: dict[str, Any] = {"accepted": [], "rejected": []}
    orig = hbp.send_peer

    def _traced(peer_id: bytes, packet: bytes, *, _skip_dual_expand: bool = False) -> None:
        route_pkt = packet
        peer = hbp._peers.get(peer_id)
        if peer is not None and packet[:4] == b"DMRD":
            from adn_server.application.routing.downlink import remap_dmrd_for_peer

            route_pkt = remap_dmrd_for_peer(packet, peer, hbp._config, peer_id=peer_id)
        opts_ok = hbp._peer_should_receive_dmrd(peer_id, packet)
        gate_ok = hbp._peer_would_accept_group_dmrd(
            peer_id, packet if not _skip_dual_expand else route_pkt, routed=_skip_dual_expand,
        )
        row = {
            "tgid": int_id(route_pkt[8:11]),
            "stream": route_pkt[16:20],
            "opts_ok": opts_ok,
            "gate_ok": gate_ok,
            "blocked": not (opts_ok and (gate_ok if packet[:4] == b"DMRD" else True)),
        }
        if row["blocked"]:
            trace["rejected"].append(row)
        else:
            trace["accepted"].append(row)
        orig(peer_id, packet, _skip_dual_expand=_skip_dual_expand)

    hbp.send_peer = _traced  # type: ignore[method-assign]
    return trace


def _inject_obp_at(
    scenario: DeterministicScenario,
    packet: bytes,
    *,
    pkt_time: float,
) -> None:
    scenario.clock.now = pkt_time
    with _patch_harness_time(scenario.clock), patch_routing_wall_time(scenario.clock):
        scenario.routing.dmrd_received(
            "OBP-CL",
            ingress_pkt_time=pkt_time,
            obp_use_parsed=True,
            obp_hops=b"\x00\x00\x00\x00",
            obp_source_server=bytes_4(9990),
            **_parse_dmrd_args(packet),
        )


def _run_interleaved_qsos(
    scenario: DeterministicScenario,
    *,
    voice_bursts: int = _VOICE_BURSTS,
    interval_s: float = _INTERVAL_S,
) -> dict[str, Any]:
    t0 = scenario.clock.time()
    step = 0
    tx_stamp_changes = 0
    prev_tx_stream: bytes | None = None
    hs_addr = ("127.0.0.1", 62031)
    delivered_a: list[bytes] = []
    delivered_b: list[bytes] = []
    blocked_a_checks: list[dict[str, Any]] = []

    def _slot_st() -> dict[str, Any]:
        proto = scenario.protocols["MASTER-A"]
        assert isinstance(proto, HBPProtocol)
        return proto.STATUS.setdefault(2, {})

    def _record_delivery(pkt: bytes) -> None:
        tgid = int_id(pkt[8:11])
        sid = pkt[16:20]
        if tgid == _TG_A and sid == bytes_4(_STREAM_A):
            delivered_a.append(pkt)
        elif tgid == _TG_B and sid == bytes_4(_STREAM_B):
            delivered_b.append(pkt)

    packets_plan: list[tuple[str, bytes]] = []
    for label, tgid, stream, rf in (
        ("A", _TG_A, _STREAM_A, _RF_A),
        ("B", _TG_B, _STREAM_B, _RF_B),
    ):
        packets_plan.append(
            (
                label,
                _dmrd_packet(
                    tgid=tgid,
                    stream_id=stream,
                    rf_src=rf,
                    frame_type=HBPF_DATA_SYNC,
                    dtype_vseq=HBPF_SLT_VHEAD,
                ),
            )
        )
    for seq in range(1, voice_bursts + 1):
        for label, tgid, stream, rf in (
            ("A", _TG_A, _STREAM_A, _RF_A),
            ("B", _TG_B, _STREAM_B, _RF_B),
        ):
            packets_plan.append(
                (
                    label,
                    _dmrd_packet(
                        tgid=tgid,
                        stream_id=stream,
                        rf_src=rf,
                        frame_type=HBPF_VOICE,
                        dtype_vseq=(seq % 4) or 1,
                        seq=seq,
                    ),
                )
            )
    for label, tgid, stream, rf in (
        ("A", _TG_A, _STREAM_A, _RF_A),
        ("B", _TG_B, _STREAM_B, _RF_B),
    ):
        packets_plan.append(
            (
                label,
                _dmrd_packet(
                    tgid=tgid,
                    stream_id=stream,
                    rf_src=rf,
                    frame_type=HBPF_DATA_SYNC,
                    dtype_vseq=HBPF_SLT_VTERM,
                    seq=99,
                ),
            )
        )

    scenario.capture.packets.clear()
    transport = scenario.protocols["MASTER-A"]
    assert isinstance(transport, HBPProtocol)
    rec = transport.transport
    assert isinstance(rec, RecordingTransport)
    rec.clear()

    hbp = transport
    hs = bytes_4(_HS_PEER)
    peer = hbp._peers[hs]

    for label, pkt in packets_plan:
        pkt_time = t0 + step * interval_s
        step += 1
        _inject_obp_at(scenario, pkt, pkt_time=pkt_time)
        cur = _slot_st().get("TX_STREAM_ID")
        if cur != prev_tx_stream:
            tx_stamp_changes += 1
            prev_tx_stream = cur
        bits = pkt[15] | 0x80
        route_pkt = pkt[:15] + bytes([bits]) + pkt[16:]
        ctx = hbp._downlink_ctx()
        if (
            label == "A"
            and int_id(pkt[8:11]) == _TG_A
            and pkt[16:20] == bytes_4(_STREAM_A)
            and (pkt[15] & 0xF) != HBPF_SLT_VTERM
        ):
            blocked = peer_slot_blocks_downlink(ctx, hs, peer, route_pkt, pkt_time=pkt_time)
            if blocked:
                blocked_a_checks.append(
                    {
                        "pkt_time": pkt_time,
                        "step": step - 1,
                        "slot_st": copy.deepcopy(_slot_st()),
                        "peer_slots": copy.deepcopy(ctx.peer_voice_slots.get(hs, {})),
                        "reason": diagnose_downlink_block(ctx, hs, peer, route_pkt, pkt_time),
                    }
                )
        for sent_pkt, addr in rec.sent:
            if addr == hs_addr:
                _record_delivery(sent_pkt)
        rec.clear()

    return {
        "tx_stamp_changes": tx_stamp_changes,
        "start_tx_events": len(
            [ev for ev in (scenario.report_factory.events if scenario.report_factory else []) if ",START,TX," in ev]
        ),
        "delivered_a": delivered_a,
        "delivered_b": delivered_b,
        "blocked_a_checks": blocked_a_checks,
        "final_slot_st": copy.deepcopy(_slot_st()),
    }


def diagnose_downlink_block(
    ctx: DownlinkContext,
    peer_id: bytes,
    peer: dict[str, Any],
    route_pkt: bytes,
    pkt_time: float,
) -> str:
    """Pin which gate branch would block this downlink (mirrors helpers/downlink)."""
    if route_pkt[:4] != b"DMRD":
        return "not_dmrd"
    stream_id = route_pkt[16:20]
    incoming_tgid_b = route_pkt[8:11]
    hang = float(ctx.sys_cfg.get("GROUP_HANGTIME", 0) or 0)
    peer_slots = ctx.peer_voice_slots.get(bytes_4(int_id(peer_id)))
    pk = bytes_4(int_id(peer_id))

    for voice_slot in (2,):
        hang_row = ctx.peer_voice_hangtime.get(pk, {}).get(voice_slot)
        slot_st = ctx.status.get(voice_slot, {})
        if _peer_transmit_hangtime_blocks(hang_row, incoming_tgid_b, pkt_time, hang):
            return f"helpers.py:_peer_transmit_hangtime_blocks voice_slot={voice_slot}"
        active = (peer_slots or {}).get(voice_slot)
        if isinstance(active, dict):
            incoming_tgid = int_id(incoming_tgid_b)
            active_tgid = int(active.get("tgid", 0) or 0)
            active_stream = active.get("stream_id")
            active_time = float(active.get("time", 0) or 0)
            age = pkt_time - active_time
            if active.get("ingress"):
                return f"helpers.py:peer_hotspot_voice_slot_busy ingress voice_slot={voice_slot}"
            if active.get("bridge_hold") and active_tgid and incoming_tgid != active_tgid:
                if age <= hang:
                    return (
                        f"helpers.py:peer_hotspot_voice_slot_busy bridge_hold "
                        f"active_tg={active_tgid} incoming={incoming_tgid}"
                    )
            if active_stream and stream_id:
                if active_stream == stream_id:
                    pass
                elif active_tgid and active_tgid == incoming_tgid:
                    if age < STREAM_TO:
                        return (
                            f"helpers.py:peer_hotspot_voice_slot_busy "
                            f"same_tg_different_stream active={active_stream!r} incoming={stream_id!r}"
                        )
                elif active_tgid and incoming_tgid != active_tgid:
                    return (
                        f"helpers.py:peer_hotspot_voice_slot_busy "
                        f"elif active_tgid={active_tgid} incoming_tgid={incoming_tgid}"
                    )
                else:
                    return "helpers.py:peer_hotspot_voice_slot_busy else branch (stream/tg mismatch)"
            elif isinstance(active, dict):
                if not (active_tgid and active_tgid == incoming_tgid):
                    return (
                        f"helpers.py:peer_hotspot_voice_slot_busy "
                        f"no_active_stream active_tg={active_tgid} incoming={incoming_tgid}"
                    )
        if peer_single_blocks_foreign_same_tg_downlink(
            peer, pk, voice_slot, incoming_tgid_b, peer_slots, ctx.sys_cfg, now=pkt_time,
        ):
            return f"helpers.py:peer_single_blocks_foreign_same_tg_downlink voice_slot={voice_slot}"
        if peer_single_same_tg_foreign_tx_blocks(
            peer, pk, incoming_tgid_b, stream_id, slot_st, ctx.sys_cfg, pkt_time=pkt_time,
        ):
            return (
                f"helpers.py:peer_single_same_tg_foreign_tx_blocks "
                f"TX_STREAM_ID={slot_st.get('TX_STREAM_ID')!r} slot_active={slot_has_active_voice(slot_st, pkt_time)}"
            )
        if bytes_4(int_id(slot_st.get("RX_PEER", b""))) == pk:
            rx_active = (
                slot_st.get("RX_TYPE") is not None
                and slot_st.get("RX_TYPE") != HBPF_SLT_VTERM
                and (pkt_time - float(slot_st.get("RX_TIME", 0))) < STREAM_TO
            )
            if rx_active and stream_id != slot_st.get("RX_STREAM_ID"):
                return (
                    f"helpers.py:peer_hotspot_voice_slot_busy "
                    f"rx_active stream_mismatch RX_STREAM_ID={slot_st.get('RX_STREAM_ID')!r}"
                )
            if _peer_status_rx_hangtime_blocks(
                peer_id, slot_st, incoming_tgid_b, pkt_time, hang,
            ):
                return f"helpers.py:_peer_status_rx_hangtime_blocks voice_slot={voice_slot}"
        if hbp_slot_blocks_group_voice_for_peer(
            slot_st,
            peer_id,
            incoming_tgid_b,
            stream_id,
            pkt_time,
            hang,
            per_peer=True,
            peers=ctx.peers,
            peer_slots=peer_slots,
            peer_hang_row=hang_row,
            voice_slot=voice_slot,
            sys_cfg=ctx.sys_cfg,
        ):
            return f"helpers.py:hbp_slot_blocks_group_voice_for_peer voice_slot={voice_slot}"
    if peer_slot_blocks_downlink(ctx, peer_id, peer, route_pkt, pkt_time=pkt_time):
        return "downlink.py:peer_slot_blocks_downlink (composite)"
    return "not_blocked"


def test_flip_flop_slot_st_blocks_active_listen_on_cross_tg() -> None:
    """Pure unit: peer listening on TG A must not RX TG B while session is open."""
    now = 1_000_000.0
    hs = bytes_4(_HS_PEER)
    peer = {"OPTIONS": f"TS2={_TG_A},{_TG_B};SINGLE=0;".encode()}
    stream_a = bytes_4(_STREAM_A)
    stream_b = bytes_4(_STREAM_B)
    peer_slots = {
        2: {"stream_id": stream_a, "tgid": _TG_A, "time": now, "ingress": False},
    }
    slot_st = {
        "TX_PEER": bytes_4(_OBP_PEER),
        "TX_STREAM_ID": stream_b,
        "TX_TYPE": HBPF_SLT_VHEAD,
        "TX_TIME": now,
        "TX_TGID": bytes_3(_TG_B),
        "RX_TYPE": HBPF_SLT_VTERM,
    }
    assert peer_hotspot_voice_slot_busy(
        hs, 2, stream_b, bytes_3(_TG_B), slot_st, peer_slots, None, now + 0.06, 5.0,
        peer=peer, sys_cfg={"GROUP_HANGTIME": 5.0, "SINGLE_MODE": False},
    )
    route_b = _dmrd_packet(
        tgid=_TG_B,
        stream_id=_STREAM_B,
        rf_src=_RF_B,
        frame_type=HBPF_VOICE,
        dtype_vseq=1,
        seq=1,
    )
    route_b = route_b[:15] + bytes([route_b[15] | 0x80]) + route_b[16:]
    ctx = DownlinkContext(
        config={"SYSTEMS": {"MASTER-A": {"MODE": "MASTER"}}},
        system_name="MASTER-A",
        sys_cfg={"GROUP_HANGTIME": 5.0, "SINGLE_MODE": False, "MODE": "MASTER"},
        peers={hs: peer},
        status={2: slot_st},
        peer_voice_slots={hs: copy.deepcopy(peer_slots)},
        connected_count=2,
    )
    reason = diagnose_downlink_block(ctx, hs, peer, route_b, now + 0.06)
    assert "active_tgid=3109050 incoming_tgid=52090" in reason


def test_flip_flop_slot_st_same_stream_not_blocked_despite_foreign_tx_stamp() -> None:
    """Same stream id on active listen must pass even when flat TX row shows other stream."""
    now = 1_000_000.0
    hs = bytes_4(_HS_PEER)
    peer = {"OPTIONS": f"TS2={_TG_A},{_TG_B};SINGLE=0;".encode()}
    stream_a = bytes_4(_STREAM_A)
    stream_b = bytes_4(_STREAM_B)
    peer_slots = {
        2: {"stream_id": stream_a, "tgid": _TG_A, "time": now, "ingress": False},
    }
    slot_st = {
        "TX_PEER": bytes_4(_OBP_PEER),
        "TX_STREAM_ID": stream_b,
        "TX_TYPE": HBPF_SLT_VHEAD,
        "TX_TIME": now,
        "TX_TGID": bytes_3(_TG_B),
        "RX_TYPE": HBPF_SLT_VTERM,
    }
    assert not peer_hotspot_voice_slot_busy(
        hs, 2, stream_a, bytes_3(_TG_A), slot_st, peer_slots, None, now + 0.06, 5.0,
        peer=peer, sys_cfg={"GROUP_HANGTIME": 5.0, "SINGLE_MODE": False},
    )


def test_rx_active_own_peer_foreign_stream_blocks_active_downlink() -> None:
    """When STATUS RX shows the hotspot mid-TX on another stream, stream-A downlink blocks.

    helpers.py peer_hotspot_voice_slot_busy lines 411-418:
    ``rx_active and stream_id != slot_st.get('RX_STREAM_ID')`` → True.
    """
    now = 1_000_000.0
    hs = bytes_4(_HS_PEER)
    peer = {"OPTIONS": f"TS2={_TG_A},{_TG_B};SINGLE=0;".encode()}
    stream_a = bytes_4(_STREAM_A)
    stream_b = bytes_4(_STREAM_B)
    peer_slots = {
        2: {"stream_id": stream_a, "tgid": _TG_A, "time": now, "ingress": False},
    }
    slot_st = {
        "RX_PEER": hs,
        "RX_STREAM_ID": stream_b,
        "RX_TYPE": HBPF_SLT_VHEAD,
        "RX_TIME": now,
        "RX_TGID": bytes_3(_TG_B),
        "TX_PEER": bytes_4(_OBP_PEER),
        "TX_STREAM_ID": stream_a,
        "TX_TYPE": HBPF_SLT_VHEAD,
        "TX_TIME": now,
        "TX_TGID": bytes_3(_TG_A),
    }
    route_a = _dmrd_packet(
        tgid=_TG_A,
        stream_id=_STREAM_A,
        rf_src=_RF_A,
        frame_type=HBPF_VOICE,
        dtype_vseq=1,
        seq=1,
    )
    route_a = route_a[:15] + bytes([route_a[15] | 0x80]) + route_a[16:]
    ctx = DownlinkContext(
        config={"SYSTEMS": {"MASTER-A": {"MODE": "MASTER"}}},
        system_name="MASTER-A",
        sys_cfg={"GROUP_HANGTIME": 5.0, "SINGLE_MODE": False, "MODE": "MASTER"},
        peers={hs: peer},
        status={2: slot_st},
        peer_voice_slots={hs: copy.deepcopy(peer_slots)},
        connected_count=2,
    )
    reason = diagnose_downlink_block(ctx, hs, peer, route_a, now + 0.06)
    assert "rx_active stream_mismatch" in reason
    assert peer_slot_blocks_downlink(ctx, hs, peer, route_a, pkt_time=now + 0.06)


def test_concurrent_obp_tx_row_flip_flop(caplog: pytest.LogCaptureFixture) -> None:
    """BUG: shared STATUS[2] TX_STREAM_ID must change at most twice per stream (VHEAD), not per packet."""
    caplog.set_level(logging.INFO)
    scenario, _hbp, _transport = _build_stack()
    result = _run_interleaved_qsos(scenario, voice_bursts=6)
    # Expect: 2 streams × 1 VHEAD stamp each (+ maybe VTERM) — not ~2×packets.
    max_expected_stamps = 6
    assert result["tx_stamp_changes"] <= max_expected_stamps, (
        f"TX_STREAM_ID flip-flop: {result['tx_stamp_changes']} changes "
        f"(expected <= {max_expected_stamps}); final STATUS[2]={result['final_slot_st']}"
    )
    assert result["start_tx_events"] <= max_expected_stamps, (
        f"START,TX report spam: {result['start_tx_events']} events"
    )


def test_concurrent_obp_active_stream_delivered_without_gaps() -> None:
    """After VHEAD, stream-A voice should not be dropped mid-QSO (minimal harness)."""
    scenario, hbp, transport = _build_stack()
    send_trace = _wrap_send_peer_trace(hbp)
    result = _run_interleaved_qsos(scenario, voice_bursts=_VOICE_BURSTS)
    voice_a = [p for p in result["delivered_a"] if _is_voice_burst(p)]
    rejected_a = [r for r in send_trace["rejected"] if r["tgid"] == _TG_A]
    expected_voice = _VOICE_BURSTS
    # Minimal OBP-only harness delivers all A packets; production drops use RX/status
    # corruption paths (see test_rx_active_own_peer_foreign_stream_blocks_active_downlink).
    assert not rejected_a, f"unexpected stream-A send_peer rejects: {rejected_a}"
    assert len(voice_a) == expected_voice, (
        f"stream A voice delivered {len(voice_a)}/{expected_voice}"
    )
  # Stream B may be dropped for the listening hotspot — not asserted.


def test_concurrent_obp_stream_b_blocked_on_listen_session() -> None:
    """Stream B must be dropped while hotspot listens to stream A (one QSO per slot)."""
    scenario, hbp, transport = _build_stack()
    send_trace = _wrap_send_peer_trace(hbp)
    _run_interleaved_qsos(scenario, voice_bursts=4)
    rejected_b = [r for r in send_trace["rejected"] if r["tgid"] == _TG_B]
    assert rejected_b, "stream B should be blocked for listening hotspot"
    ctx = hbp._downlink_ctx()
    hs = bytes_4(_HS_PEER)
    peer = hbp._peers[hs]
    sample = _dmrd_packet(
        tgid=_TG_B, stream_id=_STREAM_B, rf_src=_RF_B,
        frame_type=HBPF_VOICE, dtype_vseq=1, seq=1,
    )
    sample = sample[:15] + bytes([sample[15] | 0x80]) + sample[16:]
    reason = diagnose_downlink_block(ctx, hs, peer, sample, scenario.clock.time())
    assert (
        "active_tgid=3109050 incoming_tgid=52090" in reason
        or "_peer_transmit_hangtime_blocks" in reason
        or "peer_hotspot" in reason
        or "peer_slot" in reason
    ), f"stream B blocked but unexpected gate: {reason}"


def test_concurrent_obp_diagnose_active_stream_block_branch() -> None:
    """Identify the exact branch that drops active stream-A packets mid-QSO."""
    scenario, hbp, transport = _build_stack()
    send_trace = _wrap_send_peer_trace(hbp)
    result = _run_interleaved_qsos(scenario, voice_bursts=_VOICE_BURSTS)
    rejected_a = [
        r for r in send_trace["rejected"]
        if r["tgid"] == _TG_A and r["stream"] == bytes_4(_STREAM_A)
    ]
    if not rejected_a and not result["blocked_a_checks"]:
        # After per-leg OBP bridge TX wiring, flat TX_STREAM_ID no longer flip-flops per packet.
        if result["tx_stamp_changes"] <= 4:
            pytest.skip("no stream-A rejects; per-leg bridge TX fix removed flat-row flip-flop")
        assert result["tx_stamp_changes"] > 4, "expected TX row flip-flop in concurrent OBP harness"
        pytest.skip("no stream-A send_peer rejects — inspect flip-flop collateral (START,TX spam)")
    reasons: dict[str, int] = {}
    for row in result["blocked_a_checks"]:
        reasons[row["reason"]] = reasons.get(row["reason"], 0) + 1
    for rej in rejected_a:
        if not rej["opts_ok"]:
            key = "udp_hbp.py:_peer_should_receive_dmrd OPTIONS/eligibility"
        elif not rej["gate_ok"]:
            key = "udp_hbp.py:_peer_would_accept_group_dmrd -> peer_slot_blocks_downlink"
        else:
            key = "unknown"
        reasons[key] = reasons.get(key, 0) + 1
    top = max(reasons, key=reasons.get)
    assert "peer_hotspot" in top or "peer_slot" in top or "send_peer" in top or "udp_hbp" in top, (
        f"unexpected block reasons: {reasons}"
    )