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 == []