NODEDC_MISSION_CORE/tests/test_xgrids_application_mqtt.py

from __future__ import annotations

import hashlib
import threading
from collections import deque
from collections.abc import Callable
from datetime import UTC, datetime
from types import SimpleNamespace
from typing import Any, cast

import paho.mqtt.client as mqtt
import pytest

from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import (
    DEVICE_INFO_RESPONSE_TOPIC,
    GET_RTK_ADVANCE_REQUEST_TOPIC,
    GET_RTK_ADVANCE_RESPONSE_TOPIC,
    MODELING_STATUS_RESPONSE_TOPIC,
    LiveDeviceControlBinding,
)
from k1link.device_plugins.xgrids_k1.protocol.application_mqtt import (
    CONTROL_HEARTBEAT_TOPIC,
    CONTROL_SUBSCRIPTION_GROUPS,
    MODELING_RESPONSE_TOPIC,
    SYSTEM_ERROR_TOPIC,
    ApplicationCommandOutcomeUnknown,
    ApplicationControlDeviceFault,
    ApplicationControlProofStale,
    ApplicationMqttDeviceStatusEvidence,
    ApplicationMqttPublishEvidence,
    ApplicationMqttResponseEvidence,
    ApplicationMqttTransportError,
    ReviewedApplicationMqttTransport,
)
from k1link.device_plugins.xgrids_k1.protocol.application_publish import (
    OneShotPublishEnvelope,
)
from k1link.device_plugins.xgrids_k1.protocol.modeling_control import (
    MODELING_STATE_BASE,
    OPENAPI_SUCCESS,
    SYSTEM_ERROR_STATE_BASE,
    ModelingAction,
    SessionState,
)
from k1link.device_plugins.xgrids_k1.protocol.modeling_safety import (
    DEVICE_STATUS_TOPIC,
    MODELING_REQUEST_TOPIC,
)

VENDOR_DEVICE_ID = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
DEVICE_SERIAL = "K1SERIAL01"
APPLICATION_KEY = "11111111-2222-3333-4444-555555555555"


class FakeReasonCode:
    def __init__(self, *, failure: bool = False) -> None:
        self.is_failure = failure


class FakeClock:
    def __init__(self) -> None:
        self.now = 100.0

    def __call__(self) -> float:
        return self.now


class FakeControlClient:
    def __init__(
        self,
        *,
        emit_exchange: bool = True,
        clock: FakeClock | None = None,
        response_topic: str = DEVICE_INFO_RESPONSE_TOPIC,
    ) -> None:
        self.on_connect: Any = None
        self.on_subscribe: Any = None
        self.on_publish: Any = None
        self.on_message: Any = None
        self.on_disconnect: Any = None
        self.emit_exchange = emit_exchange
        self.clock = clock
        self.response_topic = response_topic
        self.events: deque[tuple[str, object]] = deque()
        self.connect_calls: list[tuple[str, int, int]] = []
        self.subscribe_calls: list[list[tuple[str, int]]] = []
        self.publish_calls: list[tuple[str, bytes, int, bool]] = []
        self.unsubscribe_calls: list[list[str]] = []
        self.disconnect_calls = 0
        self.next_mid = 20
        self.connect_timeout = 5.0

    def connect(self, host: str, port: int, keepalive: int) -> mqtt.MQTTErrorCode:
        self.connect_calls.append((host, port, keepalive))
        self.events.append(("connect", FakeReasonCode()))
        return mqtt.MQTT_ERR_SUCCESS

    def subscribe(self, topics: list[tuple[str, int]]) -> tuple[mqtt.MQTTErrorCode, int]:
        self.subscribe_calls.append(topics)
        mid = 6 + len(self.subscribe_calls)
        self.events.append(("subscribe", mid))
        return mqtt.MQTT_ERR_SUCCESS, mid

    def publish(
        self,
        topic: str,
        payload: bytes,
        qos: int,
        retain: bool,
    ) -> SimpleNamespace:
        self.publish_calls.append((topic, payload, qos, retain))
        mid = self.next_mid
        self.next_mid += 1
        if self.emit_exchange:
            self.events.append(("publish", mid))
            self.events.append(("message", (self.response_topic, payload)))
        return SimpleNamespace(rc=mqtt.MQTT_ERR_SUCCESS, mid=mid)

    def loop(self, timeout: float) -> mqtt.MQTTErrorCode:
        del timeout
        if self.clock is not None:
            self.clock.now += 1.0
        if not self.events:
            return mqtt.MQTT_ERR_SUCCESS
        kind, value = self.events.popleft()
        if kind == "connect":
            self.on_connect(self, None, SimpleNamespace(), value, None)
        elif kind == "subscribe":
            group = CONTROL_SUBSCRIPTION_GROUPS[int(value) - 7]
            self.on_subscribe(
                self,
                None,
                value,
                [FakeReasonCode() for _item in group],
                None,
            )
        elif kind == "publish":
            self.on_publish(self, None, value, FakeReasonCode(), None)
        elif kind == "message":
            message_parts = cast(tuple[object, ...], value)
            topic = cast(str, message_parts[0])
            payload = cast(bytes, message_parts[1])
            retained = bool(message_parts[2]) if len(message_parts) == 3 else False
            self.on_message(
                self,
                None,
                SimpleNamespace(topic=topic, payload=payload, retain=retained),
            )
        return mqtt.MQTT_ERR_SUCCESS

    def unsubscribe(self, topics: list[str]) -> tuple[mqtt.MQTTErrorCode, int]:
        self.unsubscribe_calls.append(topics)
        return mqtt.MQTT_ERR_SUCCESS, 50

    def disconnect(self) -> mqtt.MQTTErrorCode:
        self.disconnect_calls += 1
        return mqtt.MQTT_ERR_SUCCESS


class RecordingEvidenceObserver:
    def __init__(self) -> None:
        self.transport: ReviewedApplicationMqttTransport | None = None
        self.events: list[tuple[str, object]] = []

    def _prove_transport_lock_is_not_held(self) -> None:
        if self.transport is not None:
            self.transport.snapshot()

    def publish_dispatching(
        self,
        evidence: ApplicationMqttPublishEvidence,
        *,
        dispatch_admission_deadline_reached: Callable[[], bool] | None = None,
        dispatch_admission_commit: Callable[[], None] | None = None,
    ) -> None:
        self._prove_transport_lock_is_not_held()
        if (
            dispatch_admission_deadline_reached is not None
            and dispatch_admission_deadline_reached()
        ):
            raise ApplicationMqttTransportError(
                "control command dispatch deadline expired before publish admission",
                reason_code="physical-command-dispatch-deadline-expired",
            )
        if dispatch_admission_commit is not None:
            dispatch_admission_commit()
        self.events.append(("dispatching", evidence))

    def publish_result(
        self,
        evidence: ApplicationMqttPublishEvidence,
        *,
        publish_call_returned: bool,
    ) -> None:
        self._prove_transport_lock_is_not_held()
        self.events.append(
            ("publish-returned" if publish_call_returned else "publish-failed", evidence)
        )

    def qos2_completed(self, evidence: ApplicationMqttPublishEvidence) -> None:
        self._prove_transport_lock_is_not_held()
        self.events.append(("qos2", evidence))

    def application_response(self, evidence: ApplicationMqttResponseEvidence) -> None:
        self._prove_transport_lock_is_not_held()
        self.events.append(("response", evidence))

    def device_status(self, evidence: ApplicationMqttDeviceStatusEvidence) -> None:
        self._prove_transport_lock_is_not_held()
        self.events.append(("status", evidence))


def _application_message(
    session_id: str,
    *,
    device_id: str | None = None,
) -> bytes:
    header = b"".join(
        (
            _text(4, device_id) if device_id is not None else b"",
            _text(5, session_id),
            _text(6, APPLICATION_KEY),
        )
    )
    return _bytes(1, header)


def _envelope(
    operation_key: str = "bootstrap:1:DeviceInfoRequest",
    *,
    topic: str = "lixel/application/request/device_info",
    session_id: str = ":DeviceInfoRequest",
    device_id: str | None = None,
) -> OneShotPublishEnvelope:
    payload = _application_message(session_id, device_id=device_id)
    return OneShotPublishEnvelope(
        operation_key=operation_key,
        topic=topic,
        payload=payload,
        payload_sha256=hashlib.sha256(payload).hexdigest(),
        payload_bytes=len(payload),
        qos=2,
        retain=False,
    )


def _modeling_status_envelope(
    operation_key: str = "dialogue:12:ModelingStatusRequest",
) -> OneShotPublishEnvelope:
    payload = _application_message(
        f"{VENDOR_DEVICE_ID}:ModelingStatusRequest",
        device_id=VENDOR_DEVICE_ID,
    )
    return OneShotPublishEnvelope(
        operation_key=operation_key,
        topic="lixel/application/request/modeling_status",
        payload=payload,
        payload_sha256=hashlib.sha256(payload).hexdigest(),
        payload_bytes=len(payload),
        qos=2,
        retain=False,
    )


def _modeling_envelope(action: ModelingAction) -> OneShotPublishEnvelope:
    payload = _application_message(
        f"{VENDOR_DEVICE_ID}:ModelingRequest",
        device_id=VENDOR_DEVICE_ID,
    ) + _uint(2, action)
    return OneShotPublishEnvelope(
        operation_key=f"modeling:{action.name.casefold()}",
        topic=MODELING_REQUEST_TOPIC,
        payload=payload,
        payload_sha256=hashlib.sha256(payload).hexdigest(),
        payload_bytes=len(payload),
        qos=2,
        retain=False,
    )


def _modeling_response(action: ModelingAction) -> bytes:
    return (
        _application_message(
            f"{VENDOR_DEVICE_ID}:ModelingRequest",
            device_id=VENDOR_DEVICE_ID,
        )
        + _uint(2, action)
        + _bytes(15, _uint(1, 0))
    )


def _varint(value: int) -> bytes:
    encoded = bytearray()
    while value > 0x7F:
        encoded.append((value & 0x7F) | 0x80)
        value >>= 7
    encoded.append(value)
    return bytes(encoded)


def _uint(number: int, value: int) -> bytes:
    return _varint(number << 3) + _varint(value)


def _bytes(number: int, value: bytes) -> bytes:
    return _varint((number << 3) | 2) + _varint(len(value)) + value


def _text(number: int, value: str) -> bytes:
    return _bytes(number, value.encode())


def _device_status(state: SessionState, *, project_bound: bool, init_ready: bool) -> bytes:
    header = _text(4, VENDOR_DEVICE_ID)
    fields = [
        _bytes(1, header),
        _uint(2, MODELING_STATE_BASE + state),
        _text(3, DEVICE_SERIAL),
    ]
    if project_bound:
        fields.append(_text(4, "project-present"))
    if init_ready:
        fields.append(_uint(6, 1))
    return b"".join(fields)


def _system_error(state: SessionState) -> bytes:
    return _bytes(15, _uint(1, SYSTEM_ERROR_STATE_BASE + state))


def _binding() -> LiveDeviceControlBinding:
    return LiveDeviceControlBinding(
        vendor_device_id=VENDOR_DEVICE_ID,
        device_serial=DEVICE_SERIAL,
        software_version="V3.0.2-20260101-release",
        system_version="V3.0.2",
        device_model="LixelKity K1",
        device_type="A4",
        is_activated=True,
    )


def test_acceptance_transport_rejects_the_k1_access_point_fallback() -> None:
    with pytest.raises(ValueError, match="not a direct-LAN"):
        ReviewedApplicationMqttTransport("192.168.56.1")


def test_acceptance_transport_admits_the_k1_ap_only_with_an_explicit_gate() -> None:
    transport = ReviewedApplicationMqttTransport(
        "192.168.56.1",
        allow_device_ap=True,
    )

    assert transport.snapshot().state == "new"


def test_retained_control_subscription_batches_remain_exact_and_separate_from_points() -> None:
    assert [len(group) for group in CONTROL_SUBSCRIPTION_GROUPS] == [9, 5, 42]
    assert CONTROL_SUBSCRIPTION_GROUPS[0][-1] == (DEVICE_INFO_RESPONSE_TOPIC, 0)
    assert ("lixel/application/response/modeling", 2) in CONTROL_SUBSCRIPTION_GROUPS[2]
    assert all(
        topic not in {"RealtimePointcloud", "lixel/application/report/lio_pcl"}
        for group in CONTROL_SUBSCRIPTION_GROUPS
        for topic, _qos in group
    )


def test_acceptance_transport_connects_once_and_completes_one_qos2_exchange() -> None:
    fake = FakeControlClient()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )

    opened = transport.open().as_dict()
    responses = transport.exchange_batch_once(
        [_envelope()],
        required_response_operation_keys={"bootstrap:1:DeviceInfoRequest"},
    )
    transport.close()

    assert fake.connect_calls == [("192.168.1.20", 1883, 60)]
    assert fake.connect_timeout == 10.0
    assert fake.subscribe_calls == [list(group) for group in CONTROL_SUBSCRIPTION_GROUPS]
    assert fake.publish_calls == [
        (
            "lixel/application/request/device_info",
            _application_message(":DeviceInfoRequest"),
            2,
            False,
        )
    ]
    assert responses == {
        "bootstrap:1:DeviceInfoRequest": _application_message(":DeviceInfoRequest")
    }
    assert opened["clean_session"] is False
    assert opened["automatic_reconnect"] is False
    assert opened["automatic_retry"] is False
    snapshot = transport.snapshot().as_dict()
    assert snapshot["state"] == "closed"
    assert snapshot["publish_attempts"] == 1
    assert snapshot["subscribe_attempts"] == 3
    assert snapshot["qos2_completions"] == 1
    assert snapshot["correlated_responses"] == 1
    assert snapshot["operation_keys_consumed"] == 1


@pytest.mark.parametrize(
    ("connect_error", "expected_reason"),
    [
        (ConnectionRefusedError("private broker address"), "mqtt_connect_rejected"),
        (TimeoutError("private broker address"), "mqtt_connection_timeout"),
    ],
)
def test_connect_socket_failure_preserves_only_reviewed_mqtt_class(
    connect_error: OSError,
    expected_reason: str,
) -> None:
    class FailingConnectClient(FakeControlClient):
        def connect(self, host: str, port: int, keepalive: int) -> mqtt.MQTTErrorCode:
            del host, port, keepalive
            raise connect_error

    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, FailingConnectClient()),
    )

    with pytest.raises(ApplicationMqttTransportError) as raised:
        transport.open()

    assert raised.value.reason_code == expected_reason
    assert "private broker address" not in str(raised.value)
    assert transport.snapshot().automatic_retry is False


def test_consumed_operation_key_can_never_be_published_again() -> None:
    fake = FakeControlClient()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    transport.open()
    transport.exchange_batch_once(
        [_envelope()],
        required_response_operation_keys={"bootstrap:1:DeviceInfoRequest"},
    )

    with pytest.raises(ApplicationCommandOutcomeUnknown, match="already consumed"):
        transport.exchange_batch_once(
            [_envelope()],
            required_response_operation_keys={"bootstrap:1:DeviceInfoRequest"},
        )

    assert len(fake.publish_calls) == 1


def test_same_topic_inflight_responses_are_routed_by_exact_session_without_retry() -> None:
    fake = FakeControlClient(response_topic=GET_RTK_ADVANCE_RESPONSE_TOPIC)
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    transport.open()
    unbound = _envelope(
        "bootstrap:3:GetRtkAdvanceRequest",
        topic=GET_RTK_ADVANCE_REQUEST_TOPIC,
        session_id=":GetRtkAdvanceRequest",
    )
    bound_session = f"{VENDOR_DEVICE_ID}:GetRtkAdvanceRequest"
    bound = _envelope(
        "bootstrap:6:GetRtkAdvanceRequest",
        topic=GET_RTK_ADVANCE_REQUEST_TOPIC,
        session_id=bound_session,
        device_id=VENDOR_DEVICE_ID,
    )

    responses = transport.exchange_batch_once(
        [unbound, bound],
        required_response_operation_keys={unbound.operation_key, bound.operation_key},
    )

    assert responses[unbound.operation_key] == _application_message(":GetRtkAdvanceRequest")
    assert responses[bound.operation_key] == _application_message(
        bound_session,
        device_id=VENDOR_DEVICE_ID,
    )
    snapshot = transport.snapshot()
    assert snapshot.publish_attempts == 2
    assert snapshot.qos2_completions == 2
    assert snapshot.correlated_responses == 2
    assert snapshot.late_known_responses == 0
    assert snapshot.automatic_retry is False
    assert len(fake.publish_calls) == 2


def test_duplicate_start_response_blocks_stop_before_stop_is_published() -> None:
    class ModelingControlClient(FakeControlClient):
        def __init__(self) -> None:
            super().__init__(
                emit_exchange=False,
                response_topic=MODELING_RESPONSE_TOPIC,
            )

        def publish(
            self,
            topic: str,
            payload: bytes,
            qos: int,
            retain: bool,
        ) -> SimpleNamespace:
            info = super().publish(topic, payload, qos, retain)
            action = ModelingAction.START if len(self.publish_calls) == 1 else ModelingAction.STOP
            self.events.append(("publish", info.mid))
            self.events.append(
                (
                    "message",
                    (MODELING_RESPONSE_TOPIC, _modeling_response(action)),
                )
            )
            return info

    fake = ModelingControlClient()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    transport.open()

    transport.exchange_batch_once(
        [_modeling_envelope(ModelingAction.START)],
        required_response_operation_keys={"modeling:start"},
    )
    fake.events.append(
        (
            "message",
            (
                MODELING_RESPONSE_TOPIC,
                _modeling_response(ModelingAction.START),
            ),
        )
    )
    with pytest.raises(ApplicationCommandOutcomeUnknown, match="duplicate application response"):
        transport.exchange_batch_once(
            [_modeling_envelope(ModelingAction.STOP)],
            required_response_operation_keys={"modeling:stop"},
        )

    snapshot = transport.snapshot()
    assert snapshot.publish_attempts == 1
    assert snapshot.correlated_responses == 1
    assert snapshot.late_known_responses == 0
    assert snapshot.state == "poisoned"
    assert len(fake.publish_calls) == 1


def test_unanswered_optional_status_does_not_steal_same_identity_required_refresh() -> None:
    class ReusedStatusClient(FakeControlClient):
        def __init__(self) -> None:
            super().__init__(
                emit_exchange=False,
                response_topic=MODELING_STATUS_RESPONSE_TOPIC,
            )

        def publish(
            self,
            topic: str,
            payload: bytes,
            qos: int,
            retain: bool,
        ) -> SimpleNamespace:
            info = super().publish(topic, payload, qos, retain)
            self.events.append(("publish", info.mid))
            if len(self.publish_calls) == 2:
                # Ordinals 12 and 14 intentionally reuse the same protocol
                # identity.  The capture has only one response after ordinal
                # 14; unanswered optional ordinal 12 must not swallow it.
                self.events.append(("message", (MODELING_STATUS_RESPONSE_TOPIC, payload)))
            return info

    fake = ReusedStatusClient()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    transport.open()
    immediate = _modeling_status_envelope()
    refresh = _modeling_status_envelope("dialogue:14:ModelingStatusRequest")

    transport.exchange_batch_once(
        [immediate],
        required_response_operation_keys=(),
    )
    responses = transport.exchange_batch_once(
        [refresh],
        required_response_operation_keys={refresh.operation_key},
    )

    assert responses[refresh.operation_key] == refresh.payload
    snapshot = transport.snapshot()
    assert snapshot.publish_attempts == 2
    assert snapshot.correlated_responses == 1
    assert snapshot.ignored_known_responses == 0
    assert snapshot.late_known_responses == 0


def test_already_arrived_optional_status_is_consumed_before_required_refresh() -> None:
    class ReusedStatusClient(FakeControlClient):
        def __init__(self) -> None:
            super().__init__(
                emit_exchange=False,
                response_topic=MODELING_STATUS_RESPONSE_TOPIC,
            )

        def publish(
            self,
            topic: str,
            payload: bytes,
            qos: int,
            retain: bool,
        ) -> SimpleNamespace:
            info = super().publish(topic, payload, qos, retain)
            self.events.append(("publish", info.mid))
            if len(self.publish_calls) == 2:
                self.events.append(("message", (MODELING_STATUS_RESPONSE_TOPIC, payload)))
            return info

    fake = ReusedStatusClient()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    transport.open()
    immediate = _modeling_status_envelope()
    refresh = _modeling_status_envelope("dialogue:14:ModelingStatusRequest")

    transport.exchange_batch_once(
        [immediate],
        required_response_operation_keys=(),
    )
    # This callback is already available before ordinal 14 is admitted.  The
    # retained socket's zero-wait service turn must consume it as ordinal 12.
    fake.events.append(("message", (MODELING_STATUS_RESPONSE_TOPIC, immediate.payload)))
    responses = transport.exchange_batch_once(
        [refresh],
        required_response_operation_keys={refresh.operation_key},
    )

    assert responses[refresh.operation_key] == refresh.payload
    snapshot = transport.snapshot()
    assert snapshot.publish_attempts == 2
    assert snapshot.correlated_responses == 1
    assert snapshot.ignored_known_responses == 1
    assert snapshot.late_known_responses == 1


def test_unbound_optional_status_remains_distinct_from_bound_required_refresh() -> None:
    class DistinctStatusClient(FakeControlClient):
        def __init__(self) -> None:
            super().__init__(
                emit_exchange=False,
                response_topic=MODELING_STATUS_RESPONSE_TOPIC,
            )

        def publish(
            self,
            topic: str,
            payload: bytes,
            qos: int,
            retain: bool,
        ) -> SimpleNamespace:
            info = super().publish(topic, payload, qos, retain)
            self.events.append(("publish", info.mid))
            if len(self.publish_calls) == 2:
                self.events.append(
                    (
                        "message",
                        (
                            MODELING_STATUS_RESPONSE_TOPIC,
                            _application_message(":ModelingStatusRequest"),
                        ),
                    )
                )
                self.events.append(("message", (MODELING_STATUS_RESPONSE_TOPIC, payload)))
            return info

    fake = DistinctStatusClient()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    transport.open()
    initial = _envelope(
        "bootstrap:2:ModelingStatusRequest",
        topic="lixel/application/request/modeling_status",
        session_id=":ModelingStatusRequest",
    )
    refresh = _modeling_status_envelope("dialogue:14:ModelingStatusRequest")

    transport.exchange_batch_once(
        [initial],
        required_response_operation_keys=(),
    )
    responses = transport.exchange_batch_once(
        [refresh],
        required_response_operation_keys={refresh.operation_key},
    )

    assert responses[refresh.operation_key] == refresh.payload
    snapshot = transport.snapshot()
    assert snapshot.publish_attempts == 2
    assert snapshot.correlated_responses == 1
    assert snapshot.ignored_known_responses == 1
    assert snapshot.late_known_responses == 1


def test_duplicate_current_application_response_fails_closed() -> None:
    class DuplicateResponseClient(FakeControlClient):
        def publish(
            self,
            topic: str,
            payload: bytes,
            qos: int,
            retain: bool,
        ) -> SimpleNamespace:
            info = super().publish(topic, payload, qos, retain)
            self.events.append(("message", (self.response_topic, payload)))
            return info

    fake = DuplicateResponseClient()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    transport.open()

    with pytest.raises(
        ApplicationCommandOutcomeUnknown,
        match="duplicate application response",
    ) as raised:
        transport.exchange_batch_once(
            [_envelope()],
            required_response_operation_keys={"bootstrap:1:DeviceInfoRequest"},
        )

    assert raised.value.reason_code == "duplicate_application_response"
    assert transport.snapshot().state == "poisoned"
    assert len(fake.publish_calls) == 1


def test_post_start_status_can_be_response_free_and_socket_is_continuously_serviced() -> None:
    clock = FakeClock()
    fake = FakeControlClient(
        clock=clock,
        response_topic=MODELING_STATUS_RESPONSE_TOPIC,
    )
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
        monotonic=clock,
    )
    transport.open()

    responses = transport.exchange_batch_once(
        [_modeling_status_envelope()],
        required_response_operation_keys=(),
    )
    transport.maintain_open_for(
        2.0,
        allowed_response_topics={MODELING_STATUS_RESPONSE_TOPIC},
    )

    assert responses == {}
    assert transport.snapshot().state == "ready"
    assert transport.snapshot().qos2_completions == 1
    assert transport.snapshot().ignored_known_responses == 1


def test_control_owner_uses_live_state_gates_and_surfaces_system_error() -> None:
    clock = FakeClock()
    fake = FakeControlClient(clock=clock)
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
        monotonic=clock,
    )
    transport.open()

    fake.events.append(
        (
            "message",
            (
                DEVICE_STATUS_TOPIC,
                _device_status(
                    SessionState.READY,
                    project_bound=False,
                    init_ready=True,
                ),
            ),
        )
    )
    transport.maintain_open_for(1.0)
    assert transport.pre_start_ready(_binding()) is False
    assert transport.standby_complete(_binding()) is False

    fake.events.append(
        (
            "message",
            (
                DEVICE_STATUS_TOPIC,
                _device_status(
                    SessionState.READY,
                    project_bound=False,
                    init_ready=False,
                ),
            ),
        )
    )
    transport.maintain_open_for(1.0)
    assert transport.pre_start_ready(_binding()) is True

    fake.events.append(
        (
            "message",
            (
                DEVICE_STATUS_TOPIC,
                _device_status(
                    SessionState.SCANNING,
                    project_bound=True,
                    init_ready=True,
                ),
            ),
        )
    )
    transport.maintain_open_for(1.0)

    assert transport.scan_initialization_complete(_binding()) is True
    assert transport.pre_start_ready(_binding()) is False
    assert transport.standby_complete(_binding()) is False

    fake.events.append(
        (
            "message",
            (SYSTEM_ERROR_TOPIC, _system_error(SessionState.ALGORITHM_ERROR)),
        )
    )
    transport.maintain_open_for(1.0)

    snapshot = transport.snapshot().as_dict()
    assert snapshot["device_status_reports"] == 3
    assert snapshot["system_error_reports"] == 1
    assert snapshot["latest_system_error_code"] == 0x32040133
    assert snapshot["latest_system_error_state"] == "algorithm_error"
    with pytest.raises(ApplicationControlDeviceFault, match="algorithm_error"):
        transport.scan_initialization_complete(_binding())


def test_post_publish_timeout_poisoned_transport_never_retries() -> None:
    clock = FakeClock()
    fake = FakeControlClient(emit_exchange=False, clock=clock)
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        exchange_timeout_seconds=2.0,
        client_factory=lambda: cast(mqtt.Client, fake),
        monotonic=clock,
    )
    transport.open()

    with pytest.raises(ApplicationCommandOutcomeUnknown, match="automatic retry is forbidden"):
        transport.exchange_batch_once(
            [_envelope()],
            required_response_operation_keys={"bootstrap:1:DeviceInfoRequest"},
        )

    assert transport.snapshot().state == "poisoned"
    assert len(fake.publish_calls) == 1
    with pytest.raises(ApplicationCommandOutcomeUnknown, match="already consumed"):
        transport.exchange_batch_once(
            [_envelope()],
            required_response_operation_keys={"bootstrap:1:DeviceInfoRequest"},
        )
    assert len(fake.publish_calls) == 1


def test_network_loop_failure_keeps_exact_paho_result_and_phase() -> None:
    class LoopFailureClient(FakeControlClient):
        def loop(self, timeout: float) -> mqtt.MQTTErrorCode:
            if self.publish_calls and not self.events:
                return mqtt.MQTT_ERR_CONN_LOST
            return super().loop(timeout)

    fake = LoopFailureClient()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    transport.open()

    with pytest.raises(
        ApplicationCommandOutcomeUnknown,
        match=r"phase=post-publish-drain, result=7: The connection was lost",
    ):
        transport.exchange_batch_once(
            [_envelope()],
            required_response_operation_keys={"bootstrap:1:DeviceInfoRequest"},
        )

    snapshot = transport.snapshot().as_dict()
    assert snapshot["state"] == "poisoned"
    assert snapshot["last_loop_result_code"] == int(mqtt.MQTT_ERR_CONN_LOST)
    assert snapshot["last_loop_result_name"] == mqtt.error_string(int(mqtt.MQTT_ERR_CONN_LOST))
    assert snapshot["last_loop_phase"] == "post-publish-drain"
    assert len(fake.publish_calls) == 1


def test_control_proof_expires_without_remote_packets_and_heartbeat_refreshes_it() -> None:
    monotonic = FakeClock()
    suspend_aware = [1_000.0]
    fake = FakeControlClient(clock=monotonic)
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
        monotonic=monotonic,
        suspend_aware_clock=lambda: suspend_aware[0],
        control_proof_ttl_seconds=2.0,
    )
    transport.open()
    transport.exchange_batch_once(
        [_envelope()],
        required_response_operation_keys={"bootstrap:1:DeviceInfoRequest"},
    )

    initial = transport.snapshot().as_dict()
    assert initial["control_proof_revision"] == 1
    assert initial["control_proof_source"] == "correlated-application-response"
    assert initial["control_proof_fresh"] is True
    transport.validate_control_proof(_binding())

    monotonic.now += 2.01
    with pytest.raises(ApplicationControlProofStale) as stale:
        transport.validate_control_proof(_binding())
    assert stale.value.reason_code == "control_proof_stale"

    fake.events.append(("message", (CONTROL_HEARTBEAT_TOPIC, b"opaque-vendor-heartbeat")))
    transport.maintain_open_for(1.0)

    refreshed = transport.snapshot().as_dict()
    assert refreshed["control_proof_revision"] == 2
    assert refreshed["control_proof_source"] == "mqtt-heartbeat"
    assert refreshed["control_proof_fresh"] is True
    transport.validate_control_proof(_binding())


def test_control_proof_uses_suspend_aware_elapsed_time() -> None:
    monotonic = FakeClock()
    suspend_aware = [5_000.0]
    fake = FakeControlClient(clock=monotonic)
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
        monotonic=monotonic,
        suspend_aware_clock=lambda: suspend_aware[0],
        control_proof_ttl_seconds=2.0,
    )
    transport.open()
    transport.exchange_batch_once(
        [_envelope()],
        required_response_operation_keys={"bootstrap:1:DeviceInfoRequest"},
    )
    monotonic_before_sleep = monotonic.now

    suspend_aware[0] += 60.0

    assert monotonic.now == monotonic_before_sleep
    assert transport.snapshot().control_proof_fresh is False
    with pytest.raises(ApplicationControlProofStale):
        transport.validate_control_proof(_binding())


def test_device_status_refresh_is_promoted_only_after_exact_identity_validation() -> None:
    monotonic = FakeClock()
    fake = FakeControlClient(clock=monotonic)
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
        monotonic=monotonic,
    )
    transport.open()
    transport.exchange_batch_once(
        [_envelope()],
        required_response_operation_keys={"bootstrap:1:DeviceInfoRequest"},
    )
    fake.events.append(
        (
            "message",
            (
                DEVICE_STATUS_TOPIC,
                _device_status(
                    SessionState.READY,
                    project_bound=False,
                    init_ready=False,
                ),
            ),
        )
    )
    transport.maintain_open_for(1.0)

    unbound = transport.snapshot().as_dict()
    assert unbound["device_status_reports"] == 1
    assert unbound["control_proof_revision"] == 1
    assert unbound["control_proof_source"] == "correlated-application-response"

    transport.validate_control_proof(_binding())

    promoted = transport.snapshot().as_dict()
    assert promoted["control_proof_revision"] == 2
    assert promoted["control_proof_source"] == "bound-device-status"


def test_evidence_observer_sees_exact_publish_order_with_immediate_qos_callback() -> None:
    class ImmediateQosClient(FakeControlClient):
        def __init__(self) -> None:
            super().__init__(emit_exchange=False, response_topic=MODELING_RESPONSE_TOPIC)

        def publish(
            self,
            topic: str,
            payload: bytes,
            qos: int,
            retain: bool,
        ) -> SimpleNamespace:
            self.publish_calls.append((topic, payload, qos, retain))
            mid = self.next_mid
            self.next_mid += 1
            # Paho is allowed to complete a publish from inside publish().  The
            # observer must still see publish-result before QoS2 completion.
            self.on_publish(self, None, mid, FakeReasonCode(), None)
            response = (
                _application_message(
                    f"{VENDOR_DEVICE_ID}:ModelingRequest",
                    device_id=VENDOR_DEVICE_ID,
                )
                + _uint(2, ModelingAction.START)
                + _bytes(15, _uint(1, OPENAPI_SUCCESS))
            )
            self.events.append(("message", (MODELING_RESPONSE_TOPIC, response)))
            return SimpleNamespace(rc=mqtt.MQTT_ERR_SUCCESS, mid=mid)

    observed_at = datetime(2026, 8, 7, 9, 30, tzinfo=UTC)
    fake = ImmediateQosClient()
    observer = RecordingEvidenceObserver()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
        utc_now=lambda: observed_at,
    )
    observer.transport = transport
    transport.install_evidence_observer(observer)
    transport.open()
    envelope = _modeling_envelope(ModelingAction.START)

    transport.exchange_batch_once(
        [envelope],
        required_response_operation_keys={envelope.operation_key},
    )

    assert [kind for kind, _evidence in observer.events] == [
        "dispatching",
        "publish-returned",
        "qos2",
        "response",
    ]
    dispatch = cast(ApplicationMqttPublishEvidence, observer.events[0][1])
    returned = cast(ApplicationMqttPublishEvidence, observer.events[1][1])
    qos2 = cast(ApplicationMqttPublishEvidence, observer.events[2][1])
    response = cast(ApplicationMqttResponseEvidence, observer.events[3][1])
    assert dispatch.operation_key == "modeling:start"
    assert dispatch.payload_sha256 == envelope.payload_sha256
    assert dispatch.packet_id is None
    assert returned.packet_id == 20
    assert qos2 == returned
    assert response.operation_key == "modeling:start"
    assert response.modeling_action == "start"
    assert response.result_code == OPENAPI_SUCCESS
    assert response.success is True
    assert response.observed_at_utc == "2026-08-07T09:30:00.000Z"


def test_publish_exception_emits_observing_failure_without_inventing_packet_or_qos() -> None:
    class FailingPublishClient(FakeControlClient):
        def publish(
            self,
            topic: str,
            payload: bytes,
            qos: int,
            retain: bool,
        ) -> SimpleNamespace:
            self.publish_calls.append((topic, payload, qos, retain))
            raise OSError("socket disappeared during publish")

    fake = FailingPublishClient()
    observer = RecordingEvidenceObserver()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    observer.transport = transport
    transport.install_evidence_observer(observer)
    transport.open()

    with pytest.raises(ApplicationCommandOutcomeUnknown, match="publish call failed"):
        transport.exchange_batch_once(
            [_modeling_envelope(ModelingAction.START)],
            required_response_operation_keys={"modeling:start"},
        )

    assert [kind for kind, _evidence in observer.events] == [
        "dispatching",
        "publish-failed",
    ]
    failure = cast(ApplicationMqttPublishEvidence, observer.events[1][1])
    assert failure.operation_key == "modeling:start"
    assert failure.packet_id is None
    assert transport.snapshot().qos2_completions == 0


def test_device_status_evidence_preserves_wire_hash_retain_identity_and_project() -> None:
    observed_at = datetime(2026, 8, 7, 10, 15, tzinfo=UTC)
    fake = FakeControlClient(emit_exchange=False)
    observer = RecordingEvidenceObserver()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
        utc_now=lambda: observed_at,
    )
    observer.transport = transport
    transport.install_evidence_observer(observer)
    transport.open()
    payload = _device_status(
        SessionState.SCANNING,
        project_bound=True,
        init_ready=True,
    )

    fake.on_message(
        fake,
        None,
        SimpleNamespace(topic=DEVICE_STATUS_TOPIC, payload=payload, retain=False),
    )

    assert [kind for kind, _evidence in observer.events] == ["status"]
    evidence = cast(ApplicationMqttDeviceStatusEvidence, observer.events[0][1])
    assert evidence.vendor_device_id_sha256 == hashlib.sha256(VENDOR_DEVICE_ID.encode()).hexdigest()
    assert evidence.device_serial_sha256 == hashlib.sha256(DEVICE_SERIAL.encode()).hexdigest()
    assert evidence.session_state == "scanning"
    assert evidence.session_state_code == MODELING_STATE_BASE + SessionState.SCANNING
    assert evidence.project_bound is True
    assert evidence.project_id_sha256 == hashlib.sha256(b"project-present").hexdigest()
    assert evidence.init_ready is True
    assert evidence.status_message_sha256 == hashlib.sha256(payload).hexdigest()
    assert evidence.mqtt_retained is False
    assert evidence.observed_at_utc == "2026-08-07T10:15:00.000Z"


def test_evidence_observer_can_only_be_installed_before_transport_open() -> None:
    fake = FakeControlClient()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    transport.open()

    with pytest.raises(ApplicationMqttTransportError) as raised:
        transport.install_evidence_observer(RecordingEvidenceObserver())

    assert raised.value.reason_code == "evidence_observer_install_too_late"


def test_qos_observer_failure_poisoning_is_processed_without_transport_lock_deadlock() -> None:
    class RaisingQosObserver(RecordingEvidenceObserver):
        def qos2_completed(self, evidence: ApplicationMqttPublishEvidence) -> None:
            super().qos2_completed(evidence)
            raise RuntimeError("ledger persistence unavailable")

    fake = FakeControlClient()
    observer = RaisingQosObserver()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    observer.transport = transport
    transport.install_evidence_observer(observer)
    transport.open()

    with pytest.raises(ApplicationCommandOutcomeUnknown, match="evidence observer failed"):
        transport.exchange_batch_once(
            [_envelope()],
            required_response_operation_keys={"bootstrap:1:DeviceInfoRequest"},
        )

    assert transport.snapshot().state == "poisoned"
    assert [kind for kind, _evidence in observer.events][:3] == [
        "dispatching",
        "publish-returned",
        "qos2",
    ]


def test_late_old_pubcomp_cannot_be_reused_as_qos_evidence_for_a_future_mid() -> None:
    fake = FakeControlClient()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    transport.open()
    first = _envelope()
    transport.exchange_batch_once(
        [first],
        required_response_operation_keys={first.operation_key},
    )
    assert len(fake.publish_calls) == 1

    # The first exchange already consumed MID 20.  A delayed duplicate arrives
    # while no publish call owns it, immediately before Paho would reuse 20.
    fake.events.append(("publish", 20))
    fake.next_mid = 20
    second = _envelope(
        "bootstrap:2:DeviceInfoRequest",
        session_id=":DeviceInfoRequest2",
    )

    with pytest.raises(ApplicationCommandOutcomeUnknown, match="unowned or duplicate PUBCOMP"):
        transport.exchange_batch_once(
            [second],
            required_response_operation_keys={second.operation_key},
        )

    assert len(fake.publish_calls) == 1
    assert transport.snapshot().qos2_completions == 1


def test_retained_device_info_response_cannot_satisfy_live_correlation() -> None:
    class RetainedResponseClient(FakeControlClient):
        def publish(
            self,
            topic: str,
            payload: bytes,
            qos: int,
            retain: bool,
        ) -> SimpleNamespace:
            info = super().publish(topic, payload, qos, retain)
            # Replace the ordinary response queued by the fake with retained
            # broker history carrying the same deterministic session identity.
            self.events.pop()
            self.events.append(("message", (self.response_topic, payload, True)))
            return info

    fake = RetainedResponseClient()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    transport.open()

    with pytest.raises(ApplicationCommandOutcomeUnknown) as raised:
        transport.exchange_batch_once(
            [_envelope()],
            required_response_operation_keys={"bootstrap:1:DeviceInfoRequest"},
        )

    assert raised.value.reason_code == "retained_application_response"
    snapshot = transport.snapshot()
    assert snapshot.correlated_responses == 0
    assert snapshot.control_proof_revision == 0
    assert snapshot.retained_application_responses_rejected == 1


def test_retained_modeling_response_cannot_acknowledge_a_physical_command() -> None:
    class RetainedModelingResponseClient(FakeControlClient):
        def __init__(self) -> None:
            super().__init__(emit_exchange=False, response_topic=MODELING_RESPONSE_TOPIC)

        def publish(
            self,
            topic: str,
            payload: bytes,
            qos: int,
            retain: bool,
        ) -> SimpleNamespace:
            info = super().publish(topic, payload, qos, retain)
            self.events.append(("publish", info.mid))
            self.events.append(
                (
                    "message",
                    (
                        MODELING_RESPONSE_TOPIC,
                        _modeling_response(ModelingAction.START),
                        True,
                    ),
                )
            )
            return info

    fake = RetainedModelingResponseClient()
    observer = RecordingEvidenceObserver()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    observer.transport = transport
    transport.install_evidence_observer(observer)
    transport.open()

    with pytest.raises(ApplicationCommandOutcomeUnknown) as raised:
        transport.exchange_batch_once(
            [_modeling_envelope(ModelingAction.START)],
            required_response_operation_keys={"modeling:start"},
        )

    assert raised.value.reason_code == "retained_application_response"
    assert "response" not in [kind for kind, _evidence in observer.events]
    assert transport.snapshot().retained_application_responses_rejected == 1


@pytest.mark.parametrize("state", [SessionState.READY, SessionState.SCANNING])
def test_retained_device_status_cannot_change_authoritative_live_state(
    state: SessionState,
) -> None:
    fake = FakeControlClient()
    observer = RecordingEvidenceObserver()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    observer.transport = transport
    transport.install_evidence_observer(observer)
    transport.open()
    transport.exchange_batch_once(
        [_envelope()],
        required_response_operation_keys={"bootstrap:1:DeviceInfoRequest"},
    )
    observer.events.clear()
    payload = _device_status(
        state,
        project_bound=state is SessionState.SCANNING,
        init_ready=state is SessionState.SCANNING,
    )

    fake.on_message(
        fake,
        None,
        SimpleNamespace(topic=DEVICE_STATUS_TOPIC, payload=payload, retain=True),
    )

    snapshot = transport.snapshot()
    assert snapshot.latest_device_session_state is None
    assert snapshot.latest_device_project_bound is None
    assert snapshot.latest_device_init_ready is None
    assert snapshot.retained_control_reports_ignored == 1
    assert observer.events == []
    assert transport.pre_start_ready(_binding()) is False
    assert transport.scan_initialization_complete(_binding()) is False
    assert transport.standby_complete(_binding()) is False


def test_retained_heartbeat_cannot_refresh_bound_control_proof() -> None:
    monotonic = FakeClock()
    fake = FakeControlClient(clock=monotonic)
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
        monotonic=monotonic,
    )
    transport.open()
    transport.exchange_batch_once(
        [_envelope()],
        required_response_operation_keys={"bootstrap:1:DeviceInfoRequest"},
    )
    before = transport.snapshot()

    fake.on_message(
        fake,
        None,
        SimpleNamespace(
            topic=CONTROL_HEARTBEAT_TOPIC,
            payload=b"retained-old-heartbeat",
            retain=True,
        ),
    )

    after = transport.snapshot()
    assert after.control_proof_revision == before.control_proof_revision
    assert after.control_proof_source == before.control_proof_source
    assert after.retained_control_reports_ignored == 1


def test_dispatch_guard_runs_before_every_packet_and_blocks_partial_batch_continuation() -> None:
    fake = FakeControlClient()
    guard_calls = 0

    def guard() -> None:
        nonlocal guard_calls
        guard_calls += 1
        if guard_calls == 2:
            raise RuntimeError("connection epoch changed")

    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    transport.install_dispatch_guard(guard)
    transport.open()
    first = _envelope()
    second = _envelope(
        "bootstrap:2:DeviceInfoRequest",
        session_id=":DeviceInfoRequest2",
    )

    with pytest.raises(ApplicationCommandOutcomeUnknown, match="dispatch guard failed"):
        transport.exchange_batch_once(
            [first, second],
            required_response_operation_keys={first.operation_key, second.operation_key},
        )

    assert guard_calls == 2
    assert len(fake.publish_calls) == 1
    assert transport.snapshot().state == "poisoned"


def test_dispatch_guard_release_covers_exact_publish_window() -> None:
    events: list[str] = []

    class WindowAwareClient(FakeControlClient):
        def publish(
            self,
            topic: str,
            payload: bytes,
            qos: int,
            retain: bool,
        ) -> SimpleNamespace:
            assert events == ["acquired"]
            events.append("publish")
            return super().publish(topic, payload, qos, retain)

    fake = WindowAwareClient()

    def guard() -> Callable[[], None]:
        events.append("acquired")
        return lambda: events.append("released")

    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    transport.install_dispatch_guard(guard)
    transport.open()
    envelope = _envelope()
    transport.exchange_batch_once(
        [envelope],
        required_response_operation_keys={envelope.operation_key},
    )

    assert events == ["acquired", "publish", "released"]


def test_dispatch_deadline_is_checked_under_guard_before_durable_mark_and_publish() -> None:
    events: list[str] = []
    fake = FakeControlClient()
    evidence = RecordingEvidenceObserver()
    deadline_checks = 0
    admission_commits = 0

    def deadline_reached() -> bool:
        nonlocal deadline_checks
        deadline_checks += 1
        # Entry and post-drain checks are open.  Expire only while the exact
        # dispatch lease is held, immediately before durable DISPATCHING.
        return deadline_checks == 3

    def commit_admission() -> None:
        nonlocal admission_commits
        admission_commits += 1

    def guard() -> Callable[[], None]:
        events.append("acquired")
        return lambda: events.append("released")

    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    evidence.transport = transport
    transport.install_evidence_observer(evidence)
    transport.install_dispatch_guard(guard)
    transport.open()
    envelope = _modeling_envelope(ModelingAction.STOP)

    with pytest.raises(ApplicationMqttTransportError) as raised:
        transport.exchange_batch_once(
            [envelope],
            required_response_operation_keys={envelope.operation_key},
            dispatch_admission_deadline_reached=deadline_reached,
            dispatch_admission_commit=commit_admission,
        )

    assert raised.value.reason_code == "physical-command-dispatch-deadline-expired"
    assert deadline_checks == 3
    assert admission_commits == 0
    assert events == ["acquired", "released"]
    assert fake.publish_calls == []
    assert evidence.events == []
    assert transport.snapshot().publish_attempts == 0


def test_failure_after_atomic_dispatch_commit_is_unknown_and_never_publishes() -> None:
    fake = FakeControlClient()
    admission_commits = 0

    class MarkThenFailObserver(RecordingEvidenceObserver):
        def publish_dispatching(
            self,
            evidence: ApplicationMqttPublishEvidence,
            *,
            dispatch_admission_deadline_reached: Callable[[], bool] | None = None,
            dispatch_admission_commit: Callable[[], None] | None = None,
        ) -> None:
            super().publish_dispatching(
                evidence,
                dispatch_admission_deadline_reached=(
                    dispatch_admission_deadline_reached
                ),
                dispatch_admission_commit=dispatch_admission_commit,
            )
            raise OSError("injected failure after durable dispatch admission")

    def commit_admission() -> None:
        nonlocal admission_commits
        admission_commits += 1

    observer = MarkThenFailObserver()
    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    observer.transport = transport
    transport.install_evidence_observer(observer)
    transport.open()
    envelope = _modeling_envelope(ModelingAction.STOP)

    with pytest.raises(ApplicationCommandOutcomeUnknown):
        transport.exchange_batch_once(
            [envelope],
            required_response_operation_keys={envelope.operation_key},
            dispatch_admission_deadline_reached=lambda: False,
            dispatch_admission_commit=commit_admission,
        )

    assert admission_commits == 1
    assert [kind for kind, _evidence in observer.events] == ["dispatching"]
    assert fake.publish_calls == []
    assert transport.snapshot().state == "poisoned"


def test_dispatch_deadline_expiring_during_preflight_never_reaches_dispatch_guard() -> None:
    expired = False
    guard_calls = 0
    fake = FakeControlClient()
    evidence = RecordingEvidenceObserver()

    class ExpiringPreflightTransport(ReviewedApplicationMqttTransport):
        def _drain_responses(  # type: ignore[override]
            self,
            pending: object,
            responses: object,
        ) -> None:
            nonlocal expired
            super()._drain_responses(pending, responses)  # type: ignore[arg-type]
            expired = True

    def guard() -> None:
        nonlocal guard_calls
        guard_calls += 1

    transport = ExpiringPreflightTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    evidence.transport = transport
    transport.install_evidence_observer(evidence)
    transport.install_dispatch_guard(guard)
    transport.open()
    envelope = _modeling_envelope(ModelingAction.STOP)

    with pytest.raises(ApplicationMqttTransportError) as raised:
        transport.exchange_batch_once(
            [envelope],
            required_response_operation_keys={envelope.operation_key},
            dispatch_admission_deadline_reached=lambda: expired,
        )

    assert raised.value.reason_code == "physical-command-dispatch-deadline-expired"
    assert guard_calls == 0
    assert fake.publish_calls == []
    assert evidence.events == []
    assert transport.snapshot().publish_attempts == 0


def test_dispatch_deadline_expiring_after_mark_preserves_published_outcome() -> None:
    class StopResponseClient(FakeControlClient):
        def __init__(self) -> None:
            super().__init__(emit_exchange=False, response_topic=MODELING_RESPONSE_TOPIC)

        def publish(
            self,
            topic: str,
            payload: bytes,
            qos: int,
            retain: bool,
        ) -> SimpleNamespace:
            info = super().publish(topic, payload, qos, retain)
            self.events.append(("publish", info.mid))
            self.events.append(
                ("message", (MODELING_RESPONSE_TOPIC, _modeling_response(ModelingAction.STOP)))
            )
            return info

    fake = StopResponseClient()
    expired = False

    class ExpiringDispatchEvidence(RecordingEvidenceObserver):
        def publish_dispatching(
            self,
            evidence: ApplicationMqttPublishEvidence,
            *,
            dispatch_admission_deadline_reached: Callable[[], bool] | None = None,
            dispatch_admission_commit: Callable[[], None] | None = None,
        ) -> None:
            nonlocal expired
            super().publish_dispatching(
                evidence,
                dispatch_admission_deadline_reached=(
                    dispatch_admission_deadline_reached
                ),
                dispatch_admission_commit=dispatch_admission_commit,
            )
            # DISPATCHING is already durable at this callback boundary.  A
            # later cutoff must not relabel the result as no-dispatch.
            expired = True

    evidence = ExpiringDispatchEvidence()
    deadline_checks = 0

    def deadline_reached() -> bool:
        nonlocal deadline_checks
        deadline_checks += 1
        return expired

    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    evidence.transport = transport
    transport.install_evidence_observer(evidence)
    transport.open()
    envelope = _modeling_envelope(ModelingAction.STOP)

    transport.exchange_batch_once(
        [envelope],
        required_response_operation_keys={envelope.operation_key},
        dispatch_admission_deadline_reached=deadline_reached,
    )

    assert deadline_checks == 3
    assert expired is True
    assert len(fake.publish_calls) == 1
    assert [kind for kind, _item in evidence.events[:2]] == [
        "dispatching",
        "publish-returned",
    ]


def test_dispatch_guard_release_runs_when_publish_outcome_is_unknown() -> None:
    events: list[str] = []

    class FailingPublishClient(FakeControlClient):
        def publish(
            self,
            topic: str,
            payload: bytes,
            qos: int,
            retain: bool,
        ) -> SimpleNamespace:
            self.publish_calls.append((topic, payload, qos, retain))
            events.append("publish-failed")
            raise OSError("socket disappeared during publish")

    fake = FailingPublishClient()

    def guard() -> Callable[[], None]:
        events.append("acquired")
        return lambda: events.append("released")

    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    transport.install_dispatch_guard(guard)
    transport.open()

    with pytest.raises(ApplicationCommandOutcomeUnknown, match="publish call failed"):
        transport.exchange_batch_once(
            [_modeling_envelope(ModelingAction.START)],
            required_response_operation_keys={"modeling:start"},
        )

    assert events == ["acquired", "publish-failed", "released"]


def test_close_after_dispatch_precheck_prevents_queued_publish_after_gate_release() -> None:
    fake = FakeControlClient()
    guard_entered = threading.Event()
    release_guard = threading.Event()
    worker_errors: list[BaseException] = []
    evidence = RecordingEvidenceObserver()

    def guard() -> Callable[[], None]:
        guard_entered.set()
        assert release_guard.wait(3.0)
        return lambda: None

    transport = ReviewedApplicationMqttTransport(
        "192.168.1.20",
        client_factory=lambda: cast(mqtt.Client, fake),
    )
    evidence.transport = transport
    transport.install_evidence_observer(evidence)
    transport.install_dispatch_guard(guard)
    transport.open()
    envelope = _modeling_envelope(ModelingAction.STOP)

    def exchange() -> None:
        try:
            transport.exchange_batch_once(
                [envelope],
                required_response_operation_keys={envelope.operation_key},
            )
        except BaseException as exc:
            worker_errors.append(exc)

    worker = threading.Thread(target=exchange, daemon=True)
    worker.start()
    assert guard_entered.wait(3.0)

    # The worker passed the batch-level ready check but has not acquired the
    # facade publish lease. Closing in this seam must be a no-future-publish
    # proof once the lease becomes available.
    transport.close()
    release_guard.set()
    worker.join(3.0)

    assert worker.is_alive() is False
    assert len(worker_errors) == 1
    assert isinstance(worker_errors[0], ApplicationMqttTransportError)
    assert worker_errors[0].reason_code == "transport_closed_before_publish"  # type: ignore[attr-defined]
    assert fake.publish_calls == []
    assert transport.snapshot().publish_attempts == 0
    assert evidence.events == []