from __future__ import annotations import asyncio from dataclasses import replace from datetime import UTC, datetime import pytest from k1link.device_plugins.xgrids_k1.connection_supervisor import ( ConnectionSupervisor, ConnectionSupervisorClosed, EndpointTarget, HostPathProbeResult, ReadOnlyConnectionMonitor, RouteClass, TcpReachabilityProbeResult, VerifiedControlEvidence, ) NOW = datetime(2026, 8, 6, 12, 0, tzinfo=UTC) TARGET = EndpointTarget("192.168.68.52") def _clock() -> datetime: return NOW def _available_path(fingerprint: str = "en0:192.168.68.20:router-a") -> HostPathProbeResult: return HostPathProbeResult( available=True, fingerprint=fingerprint, interface="en0", source_ipv4="192.168.68.20", route_class="direct", kernel_route_fingerprint=fingerprint, ) def _unavailable_path(reason: str = "route-unavailable") -> HostPathProbeResult: return HostPathProbeResult( available=False, fingerprint=None, interface=None, source_ipv4=None, route_class="unavailable", reason_code=reason, ) def _apply_device_network( supervisor: ConnectionSupervisor, *, intent_id: str, connection_mode: str = "bridge", target: EndpointTarget = TARGET, ) -> None: assert supervisor.observe_device_network_applied( intent_id=intent_id, transport_ref="ble-k1-a", connection_mode=connection_mode, # type: ignore[arg-type] target=target, source="ble-post-write-status", ) def _reachable_supervisor() -> tuple[ConnectionSupervisor, int]: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent( intent_id="bridge-1", requested_mode="bridge", expected_device_id="k1-device-a", ) _apply_device_network(supervisor, intent_id="bridge-1") epoch = supervisor.observe_host_path(_available_path()) assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-1", host_path_epoch=epoch, reachable=True, ) assert supervisor.observe_control_evidence( VerifiedControlEvidence( intent_id="bridge-1", transport_ref="ble-k1-a", host_path_epoch=epoch, target=TARGET, connection_mode="bridge", logical_device_id="k1-device-a", compatibility_profile_id="xgrids-k1-fw-3.0.2", control_session_id="control-session-a", ) ) return supervisor, epoch def _configured_unverified_supervisor() -> tuple[ConnectionSupervisor, int]: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent( intent_id="bridge-1", requested_mode="bridge", expected_device_id="k1-device-a", ) _apply_device_network(supervisor, intent_id="bridge-1") epoch = supervisor.observe_host_path(_available_path()) assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-1", host_path_epoch=epoch, reachable=True, ) assert supervisor.snapshot().lease.state == "configured-unverified" return supervisor, epoch def test_initial_snapshot_is_fail_closed_and_exposes_only_safe_actions() -> None: snapshot = ConnectionSupervisor(clock=_clock).snapshot() assert snapshot.lease.state == "absent" assert snapshot.authority.network_mutation_allowed is False assert snapshot.authority.control_allowed is False assert snapshot.authority.acquisition_start_allowed is False assert snapshot.authority.data_ingest_authoritative is False assert snapshot.authority.physical_motion_allowed is False assert snapshot.diagnostics == () assert snapshot.allowed_actions == ("select-connection-intent",) assert snapshot.as_dict()["schema_version"] == ("missioncore.k1-connection-supervisor/v1") def test_tcp_reachability_alone_never_promotes_identity_or_authority() -> None: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent(intent_id="quick-1", requested_mode="quick-connect") epoch = supervisor.observe_host_path(_available_path()) accepted = supervisor.observe_endpoint( target=TARGET, intent_id="quick-1", host_path_epoch=epoch, reachable=True, ) snapshot = supervisor.snapshot() assert accepted is True assert snapshot.endpoint.tcp_state == "reachable" assert snapshot.device_identity.state == "unverified" assert snapshot.lease.state == "absent" assert snapshot.authority.control_allowed is False assert "verify-control-device-info" not in snapshot.allowed_actions assert "start-acquisition" not in snapshot.allowed_actions _apply_device_network( supervisor, intent_id="quick-1", connection_mode="quick-connect", ) epoch = supervisor.observe_host_path(_available_path()) assert supervisor.observe_endpoint( target=TARGET, intent_id="quick-1", host_path_epoch=epoch, reachable=True, ) configured = supervisor.snapshot() assert configured.lease.state == "configured-unverified" assert "verify-control-device-info" in configured.allowed_actions def test_exact_device_info_and_control_evidence_promote_bounded_authority() -> None: supervisor, epoch = _reachable_supervisor() snapshot = supervisor.snapshot() assert snapshot.device_identity.state == "verified" assert snapshot.control_plane.state == "healthy" assert snapshot.lease.state == "reachable" assert snapshot.lease.host_path_epoch == epoch assert snapshot.authority.control_allowed is True assert snapshot.authority.acquisition_start_allowed is True assert snapshot.authority.data_ingest_authoritative is False assert snapshot.authority.network_mutation_allowed is False assert snapshot.authority.physical_motion_allowed is False assert "start-acquisition" in snapshot.allowed_actions assert snapshot.last_known is not None assert snapshot.last_known.logical_device_id == "k1-device-a" def test_identity_mismatch_is_observed_but_never_authoritative() -> None: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent( intent_id="bridge-1", requested_mode="bridge", expected_device_id="expected-k1", ) _apply_device_network(supervisor, intent_id="bridge-1") epoch = supervisor.observe_host_path(_available_path()) supervisor.observe_endpoint( target=TARGET, intent_id="bridge-1", host_path_epoch=epoch, reachable=True, ) assert supervisor.observe_control_evidence( VerifiedControlEvidence( intent_id="bridge-1", transport_ref="ble-k1-a", host_path_epoch=epoch, target=TARGET, connection_mode="bridge", logical_device_id="different-k1", compatibility_profile_id="xgrids-k1-fw-3.0.2", control_session_id="wrong-control", ) ) snapshot = supervisor.snapshot() assert snapshot.device_identity.state == "mismatch" assert snapshot.control_plane.state == "lost" assert snapshot.lease.state == "configured-unverified" assert snapshot.authority.control_allowed is False assert "device-identity-mismatch" in snapshot.authority.reason_codes def test_host_path_epoch_change_revokes_lease_but_preserves_last_known() -> None: supervisor, first_epoch = _reachable_supervisor() second_epoch = supervisor.observe_host_path(_available_path("en0:192.168.77.20:router-b")) snapshot = supervisor.snapshot() assert second_epoch == first_epoch + 1 assert snapshot.host_path.epoch == second_epoch assert snapshot.endpoint.tcp_state == "unknown" assert snapshot.device_identity.state == "stale" assert snapshot.control_plane.state == "lost" assert snapshot.lease.state == "configured-unverified" assert snapshot.authority.control_allowed is False assert snapshot.authority.acquisition_start_allowed is False assert snapshot.last_known is not None assert snapshot.last_known.target == TARGET assert "probe-endpoint" in snapshot.allowed_actions def test_host_path_failure_and_tcp_failure_each_revoke_authority() -> None: host_failure, _ = _reachable_supervisor() failed_epoch = host_failure.observe_host_path(_unavailable_path("wifi-interface-down")) failed_snapshot = host_failure.snapshot() assert failed_epoch == 2 assert failed_snapshot.lease.state == "configured-unverified" assert failed_snapshot.authority.control_allowed is False assert failed_snapshot.allowed_actions == ( "select-connection-intent", "inspect-host-network", ) tcp_failure, epoch = _reachable_supervisor() assert tcp_failure.observe_endpoint( target=TARGET, intent_id="bridge-1", host_path_epoch=epoch, reachable=False, reason_code="mqtt-port-unreachable", ) tcp_snapshot = tcp_failure.snapshot() assert tcp_snapshot.endpoint.tcp_state == "unreachable" assert tcp_snapshot.lease.state == "configured-unverified" assert tcp_snapshot.authority.control_allowed is False assert "probe-endpoint" in tcp_snapshot.allowed_actions def test_stale_async_probe_result_cannot_restore_a_new_host_epoch() -> None: supervisor, first_epoch = _reachable_supervisor() second_epoch = supervisor.observe_host_path(_available_path("en0:192.168.99.20:router-c")) accepted = supervisor.observe_endpoint( target=TARGET, intent_id="bridge-1", host_path_epoch=first_epoch, reachable=True, ) snapshot = supervisor.snapshot() assert second_epoch == first_epoch + 1 assert accepted is False assert snapshot.endpoint.tcp_state == "unknown" assert snapshot.lease.state == "configured-unverified" assert snapshot.authority.control_allowed is False def test_quick_to_bridge_revokes_authority_with_same_target_and_host_epoch() -> None: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent( intent_id="quick-1", requested_mode="quick-connect", expected_device_id="k1-device-a", ) _apply_device_network( supervisor, intent_id="quick-1", connection_mode="quick-connect", ) epoch = supervisor.observe_host_path(_available_path()) assert supervisor.observe_endpoint( target=TARGET, intent_id="quick-1", host_path_epoch=epoch, reachable=True, ) assert supervisor.observe_control_evidence( VerifiedControlEvidence( intent_id="quick-1", transport_ref="ble-k1-a", host_path_epoch=epoch, target=TARGET, connection_mode="quick-connect", logical_device_id="k1-device-a", compatibility_profile_id="xgrids-k1-fw-3.0.2", control_session_id="quick-control", ) ) assert supervisor.snapshot().authority.control_allowed is True switched = supervisor.set_intent( intent_id="bridge-2", requested_mode="bridge", expected_device_id="k1-device-a", ) assert switched.host_path.epoch == epoch assert switched.endpoint.target is None assert switched.endpoint.intent_id == "bridge-2" assert switched.device_identity.state == "unverified" assert switched.control_plane.state == "lost" assert switched.lease.state == "lost" assert switched.last_known is not None assert switched.last_known.target == TARGET assert switched.authority.control_allowed is False _apply_device_network(supervisor, intent_id="bridge-2") assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-2", host_path_epoch=epoch, reachable=True, ) assert supervisor.observe_control_evidence( VerifiedControlEvidence( intent_id="bridge-2", transport_ref="ble-k1-a", host_path_epoch=epoch, target=TARGET, connection_mode="bridge", logical_device_id="k1-device-a", compatibility_profile_id="xgrids-k1-fw-3.0.2", control_session_id="bridge-control", ) ) restored = supervisor.snapshot() assert restored.host_path.epoch == epoch assert restored.lease.intent_id == "bridge-2" assert restored.lease.connection_mode == "bridge" assert restored.authority.control_allowed is True def test_stale_old_intent_tcp_and_control_evidence_are_rejected() -> None: supervisor, epoch = _reachable_supervisor() supervisor.set_intent( intent_id="bridge-2", requested_mode="bridge", expected_device_id="k1-device-a", ) assert ( supervisor.observe_endpoint( target=TARGET, intent_id="bridge-1", host_path_epoch=epoch, reachable=True, ) is False ) assert ( supervisor.observe_control_evidence( VerifiedControlEvidence( intent_id="bridge-1", transport_ref="ble-k1-a", host_path_epoch=epoch, target=TARGET, connection_mode="bridge", logical_device_id="k1-device-a", compatibility_profile_id="xgrids-k1-fw-3.0.2", control_session_id="stale-control", ) ) is False ) snapshot = supervisor.snapshot() assert snapshot.intent is not None assert snapshot.intent.intent_id == "bridge-2" assert snapshot.endpoint.tcp_state == "unknown" assert snapshot.device_identity.state == "unverified" assert snapshot.lease.state == "lost" assert snapshot.authority.control_allowed is False def test_device_info_from_a_different_ble_transport_is_rejected() -> None: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent( intent_id="bridge-transport-bound", requested_mode="bridge", expected_device_id="k1-device-a", ) _apply_device_network(supervisor, intent_id="bridge-transport-bound") epoch = supervisor.observe_host_path(_available_path()) assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-transport-bound", host_path_epoch=epoch, reachable=True, ) accepted = supervisor.observe_control_evidence( VerifiedControlEvidence( intent_id="bridge-transport-bound", transport_ref="ble-k1-b", host_path_epoch=epoch, target=TARGET, connection_mode="bridge", logical_device_id="k1-device-a", compatibility_profile_id="xgrids-k1-fw-3.0.2", control_session_id="wrong-transport-control", ) ) assert accepted is False snapshot = supervisor.snapshot() assert snapshot.device_network.transport_ref == "ble-k1-a" assert snapshot.device_identity.state == "unverified" assert snapshot.control_plane.state != "healthy" assert snapshot.authority.control_allowed is False @pytest.mark.parametrize("route_class", ["default", "tunnel"]) def test_non_direct_route_never_grants_control_authority(route_class: RouteClass) -> None: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent( intent_id="bridge-1", requested_mode="bridge", expected_device_id="k1-device-a", ) _apply_device_network(supervisor, intent_id="bridge-1") epoch = supervisor.observe_host_path( HostPathProbeResult( available=True, fingerprint=f"en0:192.168.68.20:{route_class}", interface="utun4" if route_class == "tunnel" else "en0", source_ipv4="192.168.68.20", route_class=route_class, ) ) assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-1", host_path_epoch=epoch, reachable=True, ) assert supervisor.observe_control_evidence( VerifiedControlEvidence( intent_id="bridge-1", transport_ref="ble-k1-a", host_path_epoch=epoch, target=TARGET, connection_mode="bridge", logical_device_id="k1-device-a", compatibility_profile_id="xgrids-k1-fw-3.0.2", control_session_id=f"{route_class}-control", ) ) snapshot = supervisor.snapshot() assert snapshot.device_identity.state == "verified" assert snapshot.lease.state == "reachable" assert snapshot.authority.control_allowed is False assert snapshot.authority.acquisition_start_allowed is False assert "host-route-not-direct" in snapshot.authority.reason_codes def test_explicit_data_expiry_revokes_only_data_authority() -> None: supervisor, epoch = _reachable_supervisor() assert supervisor.observe_data_plane( intent_id="bridge-1", host_path_epoch=epoch, state="healthy", session_id="data-session-a", ) healthy = supervisor.snapshot() assert healthy.authority.control_allowed is True assert healthy.authority.data_ingest_authoritative is True assert "stop-acquisition" in healthy.allowed_actions assert supervisor.observe_data_plane( intent_id="bridge-1", host_path_epoch=epoch, state="lost", session_id="data-session-a", reason_code="data-producer-observation-expired", ) data_lost = supervisor.snapshot() assert data_lost.lease.state == "reachable" assert data_lost.control_plane.state == "healthy" assert data_lost.authority.control_allowed is True assert data_lost.authority.acquisition_start_allowed is False assert data_lost.authority.data_ingest_authoritative is False assert "acknowledge-data-loss" in data_lost.allowed_actions def test_mqtt_control_loss_preserves_fresh_data_session_as_evidence() -> None: supervisor, epoch = _reachable_supervisor() assert supervisor.observe_data_plane( intent_id="bridge-1", host_path_epoch=epoch, state="healthy", session_id="data-session-generation-7", ) assert supervisor.observe_control_loss( intent_id="bridge-1", host_path_epoch=epoch, control_session_id="control-session-a", reason_code="mqtt-control-loop-lost", ) control_lost = supervisor.snapshot() assert control_lost.lease.state == "configured-unverified" assert control_lost.authority.control_allowed is False assert control_lost.authority.acquisition_start_allowed is False assert control_lost.authority.data_ingest_authoritative is False assert control_lost.data_plane.state == "healthy" assert control_lost.data_plane.session_id == "data-session-generation-7" assert control_lost.data_plane.host_path_epoch == epoch assert [item.code for item in control_lost.diagnostics] == ["host.mqtt.transport-unavailable"] # A packet from the already-bound producer can refresh data evidence even # though control is unavailable. It does not restore command authority. assert supervisor.observe_data_plane( intent_id="bridge-1", host_path_epoch=epoch, state="healthy", session_id="data-session-generation-7", ) refreshed = supervisor.snapshot() assert refreshed.data_plane.state == "healthy" assert refreshed.authority.control_allowed is False assert refreshed.authority.data_ingest_authoritative is False def test_k1_reboot_same_ip_rejects_phantom_control_snapshot_until_fresh_session() -> None: supervisor, epoch = _reachable_supervisor() stale_evidence = VerifiedControlEvidence( intent_id="bridge-1", transport_ref="ble-k1-a", host_path_epoch=epoch, target=TARGET, connection_mode="bridge", logical_device_id="k1-device-a", compatibility_profile_id="xgrids-k1-fw-3.0.2", control_session_id="control-session-a", control_proof_revision=1, ) assert supervisor.observe_control_loss( intent_id="bridge-1", host_path_epoch=epoch, control_session_id="control-session-a", reason_code="k1-reboot-control-loop-lost", ) assert supervisor.observe_control_evidence(stale_evidence) is False rebooted = supervisor.snapshot() assert rebooted.host_path.epoch == epoch assert rebooted.endpoint.tcp_state == "reachable" assert rebooted.control_plane.state == "lost" assert rebooted.lease.state == "configured-unverified" assert rebooted.authority.control_allowed is False assert rebooted.authority.acquisition_start_allowed is False assert supervisor.observe_control_evidence( replace( stale_evidence, control_session_id="control-session-after-reboot", ) ) recovered = supervisor.snapshot() assert recovered.control_plane.state == "healthy" assert recovered.control_plane.session_id == "control-session-after-reboot" assert recovered.authority.control_allowed is True def test_host_route_loss_preserves_fresh_data_on_its_original_epoch() -> None: supervisor, epoch = _reachable_supervisor() assert supervisor.observe_data_plane( intent_id="bridge-1", host_path_epoch=epoch, state="healthy", session_id="data-session-generation-8", ) next_epoch = supervisor.observe_host_path(_unavailable_path("router-lost")) lost = supervisor.snapshot() assert next_epoch == epoch + 1 assert lost.host_path.available is False assert lost.control_plane.state == "lost" assert lost.lease.state == "configured-unverified" assert lost.authority.control_allowed is False assert lost.authority.acquisition_start_allowed is False assert lost.authority.data_ingest_authoritative is False assert lost.data_plane.state == "healthy" assert lost.data_plane.session_id == "data-session-generation-8" assert lost.data_plane.host_path_epoch == epoch assert [item.code for item in lost.diagnostics] == ["host.route.unavailable"] # The old bounded session remains admissible as evidence, but cannot be # rebound to the replacement host epoch by a late producer callback. assert supervisor.observe_data_plane( intent_id="bridge-1", host_path_epoch=epoch, state="healthy", session_id="data-session-generation-8", ) assert ( supervisor.observe_data_plane( intent_id="bridge-1", host_path_epoch=next_epoch, state="healthy", session_id="data-session-generation-8", ) is False ) def test_tcp_loss_preserves_fresh_data_session_as_evidence() -> None: supervisor, epoch = _reachable_supervisor() assert supervisor.observe_data_plane( intent_id="bridge-1", host_path_epoch=epoch, state="healthy", session_id="data-session-generation-9", ) assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-1", host_path_epoch=epoch, reachable=False, reason_code="tcp-route-lost", ) lost = supervisor.snapshot() assert lost.endpoint.tcp_state == "unreachable" assert lost.control_plane.state == "lost" assert lost.lease.state == "configured-unverified" assert lost.authority.control_allowed is False assert lost.authority.acquisition_start_allowed is False assert lost.authority.data_ingest_authoritative is False assert lost.data_plane.state == "healthy" assert lost.data_plane.session_id == "data-session-generation-9" assert lost.data_plane.host_path_epoch == epoch assert [item.code for item in lost.diagnostics] == ["host.tcp.endpoint-unavailable"] def test_late_control_loss_from_old_intent_or_session_cannot_revoke_new_control() -> None: supervisor, epoch = _reachable_supervisor() supervisor.set_intent( intent_id="bridge-2", requested_mode="bridge", expected_device_id="k1-device-a", ) _apply_device_network(supervisor, intent_id="bridge-2") assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-2", host_path_epoch=epoch, reachable=True, ) assert supervisor.observe_control_evidence( VerifiedControlEvidence( intent_id="bridge-2", transport_ref="ble-k1-a", host_path_epoch=epoch, target=TARGET, connection_mode="bridge", logical_device_id="k1-device-a", compatibility_profile_id="xgrids-k1-fw-3.0.2", control_session_id="control-session-b", ) ) assert ( supervisor.observe_control_loss( intent_id="bridge-1", host_path_epoch=epoch, control_session_id="control-session-a", reason_code="late-old-worker-loss", ) is False ) assert ( supervisor.observe_control_loss( intent_id="bridge-2", host_path_epoch=epoch, control_session_id="control-session-a", reason_code="late-old-session-loss", ) is False ) snapshot = supervisor.snapshot() assert snapshot.intent is not None assert snapshot.intent.intent_id == "bridge-2" assert snapshot.control_plane.state == "healthy" assert snapshot.control_plane.session_id == "control-session-b" assert snapshot.lease.state == "reachable" assert snapshot.authority.control_allowed is True def test_late_data_loss_from_old_session_cannot_poison_new_data_session() -> None: supervisor, epoch = _reachable_supervisor() assert supervisor.observe_data_plane( intent_id="bridge-1", host_path_epoch=epoch, state="healthy", session_id="data-session-old", ) assert supervisor.observe_data_plane( intent_id="bridge-1", host_path_epoch=epoch, state="healthy", session_id="data-session-new", ) assert ( supervisor.observe_data_plane( intent_id="bridge-1", host_path_epoch=epoch, state="lost", session_id="data-session-old", reason_code="late-old-data-worker-loss", ) is False ) snapshot = supervisor.snapshot() assert snapshot.data_plane.state == "healthy" assert snapshot.data_plane.session_id == "data-session-new" assert snapshot.authority.control_allowed is True assert snapshot.authority.data_ingest_authoritative is True def test_monitor_injections_are_read_only_and_drop_delayed_tcp_result() -> None: async def scenario() -> None: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent(intent_id="bridge-1", requested_mode="bridge") _apply_device_network(supervisor, intent_id="bridge-1") tcp_started = asyncio.Event() release_tcp = asyncio.Event() tcp_calls: list[EndpointTarget] = [] async def host_probe(target: EndpointTarget) -> HostPathProbeResult: assert target == TARGET return _available_path() async def tcp_probe(target: EndpointTarget) -> bool: tcp_calls.append(target) tcp_started.set() await release_tcp.wait() return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) poll = asyncio.create_task(monitor.poll_once()) await tcp_started.wait() supervisor.observe_host_path(_available_path("en0:192.168.77.20:router-b")) release_tcp.set() await poll snapshot = supervisor.snapshot() assert tcp_calls == [TARGET] assert snapshot.host_path.epoch == 2 assert snapshot.endpoint.tcp_state == "unknown" assert snapshot.authority.control_allowed is False await monitor.close() assert supervisor.snapshot().closed is True with pytest.raises(ConnectionSupervisorClosed): await monitor.poll_once() asyncio.run(scenario()) def test_monitor_host_path_cas_rejects_switch_after_target_check() -> None: async def scenario() -> None: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent(intent_id="bridge-a", requested_mode="bridge") _apply_device_network(supervisor, intent_id="bridge-a") next_target = EndpointTarget("192.168.68.77") tcp_calls = 0 async def host_probe(_target: EndpointTarget) -> HostPathProbeResult: return _available_path() async def tcp_probe(_target: EndpointTarget) -> bool: nonlocal tcp_calls tcp_calls += 1 return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) def switch_after_check( expected: EndpointTarget, *, expected_intent_id: str, ) -> bool: assert expected == TARGET assert expected_intent_id == "bridge-a" supervisor.set_intent(intent_id="bridge-b", requested_mode="bridge") _apply_device_network( supervisor, intent_id="bridge-b", target=next_target, ) # Reproduce the exact old boolean-check -> reducer-write gap. return True monitor._target_is_current = switch_after_check # type: ignore[method-assign] # noqa: SLF001 snapshot = await monitor.poll_once() assert tcp_calls == 0 assert snapshot.intent is not None assert snapshot.intent.intent_id == "bridge-b" assert snapshot.device_network.target == next_target assert snapshot.host_path.epoch == 0 assert snapshot.host_path.available is False assert snapshot.host_path.reason_code == "host-path-not-observed" await monitor.close() asyncio.run(scenario()) def test_monitor_failure_cas_rejects_switch_after_target_check() -> None: async def scenario() -> None: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent(intent_id="bridge-a", requested_mode="bridge") _apply_device_network(supervisor, intent_id="bridge-a") next_target = EndpointTarget("192.168.68.77") async def host_probe(_target: EndpointTarget) -> HostPathProbeResult: raise OSError("old-target host failure") async def tcp_probe(_target: EndpointTarget) -> bool: raise AssertionError("TCP must not run after a host failure") monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) def switch_after_check( expected: EndpointTarget, *, expected_intent_id: str, ) -> bool: assert expected == TARGET assert expected_intent_id == "bridge-a" supervisor.set_intent(intent_id="bridge-b", requested_mode="bridge") _apply_device_network( supervisor, intent_id="bridge-b", target=next_target, ) return True monitor._target_is_current = switch_after_check # type: ignore[method-assign] # noqa: SLF001 snapshot = await monitor.poll_once() assert snapshot.intent is not None assert snapshot.intent.intent_id == "bridge-b" assert snapshot.device_network.target == next_target assert snapshot.host_path.epoch == 0 assert snapshot.host_path.available is False assert snapshot.host_path.reason_code == "host-path-not-observed" await monitor.close() asyncio.run(scenario()) def test_monitor_target_provider_error_cannot_poison_a_new_same_target_intent() -> None: async def scenario() -> None: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent(intent_id="bridge-a", requested_mode="bridge") _apply_device_network(supervisor, intent_id="bridge-a") target_checks = 0 async def host_probe(_target: EndpointTarget) -> HostPathProbeResult: raise OSError("old-intent host failure") async def tcp_probe(_target: EndpointTarget) -> bool: raise AssertionError("TCP must not run after a host failure") def target_provider() -> EndpointTarget: nonlocal target_checks target_checks += 1 if target_checks == 1: return TARGET # The next intent legitimately reuses the same host address. A # provider error from A must still be rejected by the intent CAS. supervisor.set_intent(intent_id="bridge-b", requested_mode="bridge") _apply_device_network(supervisor, intent_id="bridge-b") raise RuntimeError("late target lookup failure from bridge-a") monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=target_provider, ) snapshot = await monitor.poll_once() assert target_checks == 2 assert snapshot.intent is not None assert snapshot.intent.intent_id == "bridge-b" assert snapshot.device_network.target == TARGET assert snapshot.host_path.epoch == 0 assert snapshot.host_path.reason_code == "host-path-not-observed" await monitor.close() asyncio.run(scenario()) @pytest.mark.parametrize("changed_final_path", [False, True]) def test_monitor_final_publication_cas_rejects_switch_after_target_check( changed_final_path: bool, ) -> None: async def scenario() -> None: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent(intent_id="bridge-a", requested_mode="bridge") _apply_device_network(supervisor, intent_id="bridge-a") next_target = EndpointTarget("192.168.68.77") first_path = _available_path() final_path = ( _available_path("en0:192.168.99.20:late-old-target-route") if changed_final_path else first_path ) paths = iter((first_path, final_path)) target_checks = 0 async def host_probe(_target: EndpointTarget) -> HostPathProbeResult: return next(paths) async def tcp_probe(_target: EndpointTarget) -> bool: return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) def switch_on_final_check( expected: EndpointTarget, *, expected_intent_id: str, ) -> bool: nonlocal target_checks assert expected == TARGET assert expected_intent_id == "bridge-a" target_checks += 1 if target_checks == 2: supervisor.set_intent(intent_id="bridge-b", requested_mode="bridge") _apply_device_network( supervisor, intent_id="bridge-b", target=next_target, ) return True monitor._target_is_current = switch_on_final_check # type: ignore[method-assign] # noqa: SLF001 snapshot = await monitor.poll_once() assert target_checks == 2 assert snapshot.intent is not None assert snapshot.intent.intent_id == "bridge-b" assert snapshot.device_network.target == next_target assert snapshot.host_path.epoch == 1 assert snapshot.host_path.fingerprint == first_path.fingerprint assert snapshot.endpoint.tcp_state == "unknown" assert snapshot.endpoint.target is None assert snapshot.authority.control_allowed is False await monitor.close() asyncio.run(scenario()) def test_monitor_drops_changed_final_path_after_target_switch() -> None: async def scenario() -> None: supervisor, initial_epoch = _reachable_supervisor() next_target = EndpointTarget("192.168.68.77") selected_target = TARGET final_probe_started = asyncio.Event() release_final_probe = asyncio.Event() host_probe_calls = 0 async def host_probe(target: EndpointTarget) -> HostPathProbeResult: nonlocal host_probe_calls assert target == TARGET host_probe_calls += 1 if host_probe_calls == 1: return _available_path() final_probe_started.set() await release_final_probe.wait() return _available_path("en0:192.168.77.20:late-old-target-route") async def tcp_probe(target: EndpointTarget) -> bool: assert target == TARGET return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: selected_target, ) poll = asyncio.create_task(monitor.poll_once()) await final_probe_started.wait() selected_target = next_target release_final_probe.set() snapshot = await poll assert host_probe_calls == 2 assert snapshot.host_path.epoch == initial_epoch assert snapshot.host_path.fingerprint == _available_path().fingerprint assert snapshot.endpoint.target == TARGET assert snapshot.endpoint.tcp_state == "reachable" assert snapshot.authority.control_allowed is True await monitor.close() asyncio.run(scenario()) def test_monitor_close_during_final_host_probe_drops_changed_path_without_error() -> None: async def scenario() -> None: supervisor, initial_epoch = _reachable_supervisor() final_probe_started = asyncio.Event() release_final_probe = asyncio.Event() host_probe_calls = 0 async def host_probe(target: EndpointTarget) -> HostPathProbeResult: nonlocal host_probe_calls assert target == TARGET host_probe_calls += 1 if host_probe_calls == 1: return _available_path() final_probe_started.set() await release_final_probe.wait() return _available_path("en0:192.168.99.20:late-closed-route") async def tcp_probe(target: EndpointTarget) -> bool: assert target == TARGET return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) poll = asyncio.create_task(monitor.poll_once()) await final_probe_started.wait() await monitor.close() release_final_probe.set() snapshot = await poll assert host_probe_calls == 2 assert snapshot.closed is True assert snapshot.host_path.epoch == initial_epoch assert snapshot.host_path.fingerprint == _available_path().fingerprint assert snapshot.authority.control_allowed is False asyncio.run(scenario()) @pytest.mark.parametrize("failing_probe_call", [1, 2]) def test_monitor_drops_host_probe_exception_after_target_switch( failing_probe_call: int, ) -> None: async def scenario() -> None: supervisor, initial_epoch = _reachable_supervisor() next_target = EndpointTarget("192.168.68.77") selected_target = TARGET failing_probe_started = asyncio.Event() release_failing_probe = asyncio.Event() host_probe_calls = 0 tcp_probe_calls = 0 async def host_probe(target: EndpointTarget) -> HostPathProbeResult: nonlocal host_probe_calls assert target == TARGET host_probe_calls += 1 if host_probe_calls != failing_probe_call: return _available_path() failing_probe_started.set() await release_failing_probe.wait() raise OSError("late host failure for old target") async def tcp_probe(target: EndpointTarget) -> bool: nonlocal tcp_probe_calls assert target == TARGET tcp_probe_calls += 1 return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: selected_target, ) poll = asyncio.create_task(monitor.poll_once()) await failing_probe_started.wait() selected_target = next_target release_failing_probe.set() snapshot = await poll assert host_probe_calls == failing_probe_call assert tcp_probe_calls == failing_probe_call - 1 assert snapshot.host_path.epoch == initial_epoch assert snapshot.host_path.available is True assert snapshot.host_path.reason_code is None assert snapshot.endpoint.target == TARGET assert snapshot.endpoint.tcp_state == "reachable" assert snapshot.authority.control_allowed is True await monitor.close() asyncio.run(scenario()) def test_monitor_close_during_final_host_probe_exception_drops_failure() -> None: async def scenario() -> None: supervisor, initial_epoch = _reachable_supervisor() final_probe_started = asyncio.Event() release_final_probe = asyncio.Event() host_probe_calls = 0 async def host_probe(target: EndpointTarget) -> HostPathProbeResult: nonlocal host_probe_calls assert target == TARGET host_probe_calls += 1 if host_probe_calls == 1: return _available_path() final_probe_started.set() await release_final_probe.wait() raise OSError("late final host failure after close") async def tcp_probe(target: EndpointTarget) -> bool: assert target == TARGET return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) poll = asyncio.create_task(monitor.poll_once()) await final_probe_started.wait() await monitor.close() release_final_probe.set() snapshot = await poll assert host_probe_calls == 2 assert snapshot.closed is True assert snapshot.host_path.epoch == initial_epoch assert snapshot.host_path.available is True assert snapshot.host_path.reason_code is None assert snapshot.authority.control_allowed is False asyncio.run(scenario()) def test_monitor_projects_typed_tcp_refusal_without_socket_details() -> None: async def scenario() -> None: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent(intent_id="bridge-refused", requested_mode="bridge") _apply_device_network(supervisor, intent_id="bridge-refused") async def host_probe(_target: EndpointTarget) -> HostPathProbeResult: return _available_path() async def tcp_probe(_target: EndpointTarget) -> TcpReachabilityProbeResult: return TcpReachabilityProbeResult( reachable=False, reason_code="tcp-connection-refused", ) monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) snapshot = await monitor.poll_once() assert snapshot.endpoint.tcp_state == "unreachable" assert snapshot.endpoint.reason_code == "tcp-connection-refused" assert [item.code for item in snapshot.diagnostics] == ["host.tcp.connection-refused"] await monitor.close() asyncio.run(scenario()) def test_monitor_skips_tcp_when_route_probe_fails_and_closes_fail_closed() -> None: async def scenario() -> None: supervisor, _ = _reachable_supervisor() tcp_calls = 0 async def host_probe(target: EndpointTarget) -> HostPathProbeResult: assert target == TARGET return _unavailable_path("router-disappeared") async def tcp_probe(target: EndpointTarget) -> bool: nonlocal tcp_calls assert target == TARGET tcp_calls += 1 return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) snapshot = await monitor.poll_once() assert tcp_calls == 0 assert snapshot.host_path.available is False assert snapshot.lease.state == "configured-unverified" assert snapshot.authority.control_allowed is False await monitor.close() asyncio.run(scenario()) def test_monitor_host_probe_exception_fails_closed_without_leaking_exception() -> None: async def scenario() -> None: supervisor, _ = _reachable_supervisor() async def host_probe(target: EndpointTarget) -> HostPathProbeResult: assert target == TARGET raise OSError("synthetic route diagnostic") async def tcp_probe(target: EndpointTarget) -> bool: assert target == TARGET raise AssertionError("TCP must not run after a host-path failure") monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) snapshot = await monitor.poll_once() assert snapshot.host_path.reason_code == "host-path-probe-error" assert [item.code for item in snapshot.diagnostics] == ["host.route.unavailable"] assert snapshot.lease.state == "configured-unverified" assert snapshot.authority.control_allowed is False await monitor.close() asyncio.run(scenario()) def test_monitor_close_during_external_probe_drops_the_late_result() -> None: async def scenario() -> None: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent(intent_id="bridge-1", requested_mode="bridge") _apply_device_network(supervisor, intent_id="bridge-1") tcp_started = asyncio.Event() release_tcp = asyncio.Event() async def host_probe(target: EndpointTarget) -> HostPathProbeResult: assert target == TARGET return _available_path() async def tcp_probe(target: EndpointTarget) -> bool: assert target == TARGET tcp_started.set() await release_tcp.wait() return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) poll = asyncio.create_task(monitor.poll_once()) await tcp_started.wait() await monitor.close() release_tcp.set() snapshot = await poll assert snapshot.closed is True assert snapshot.endpoint.tcp_state == "unknown" assert snapshot.authority.control_allowed is False asyncio.run(scenario()) def test_monitor_close_rejects_a_poll_queued_behind_an_inflight_probe() -> None: async def scenario() -> None: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent(intent_id="bridge-queued-close", requested_mode="bridge") _apply_device_network(supervisor, intent_id="bridge-queued-close") first_probe_started = asyncio.Event() release_first_probe = asyncio.Event() host_calls = 0 async def host_probe(target: EndpointTarget) -> HostPathProbeResult: nonlocal host_calls assert target == TARGET host_calls += 1 if host_calls > 1: raise AssertionError("a queued poll contacted the host after close") first_probe_started.set() await release_first_probe.wait() return _available_path() async def tcp_probe(_target: EndpointTarget) -> bool: raise AssertionError("close must supersede TCP contact") monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) first_poll = asyncio.create_task(monitor.poll_once()) await first_probe_started.wait() queued_poll = asyncio.create_task(monitor.poll_once()) # Let the queued poll pass the pre-lock closed check and block on the # first poll's serialization lock before closing the monitor. await asyncio.sleep(0) await monitor.close() release_first_probe.set() first_snapshot = await first_poll assert first_snapshot.closed is True with pytest.raises(ConnectionSupervisorClosed): await queued_poll assert host_calls == 1 asyncio.run(scenario()) def test_monitor_loop_recovers_after_one_unexpected_iteration_failure() -> None: async def scenario() -> None: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent(intent_id="bridge-recovery", requested_mode="bridge") _apply_device_network(supervisor, intent_id="bridge-recovery") recovered = asyncio.Event() poll_calls = 0 async def host_probe(target: EndpointTarget) -> HostPathProbeResult: assert target == TARGET return _available_path() async def tcp_probe(target: EndpointTarget) -> bool: assert target == TARGET return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, interval_seconds=0.05, ) original_poll_once = monitor.poll_once async def fail_once_then_poll() -> object: nonlocal poll_calls poll_calls += 1 if poll_calls == 1: raise RuntimeError("synthetic monitor iteration failure") snapshot = await original_poll_once() recovered.set() return snapshot monitor.poll_once = fail_once_then_poll # type: ignore[method-assign] await monitor.start() await asyncio.wait_for(recovered.wait(), timeout=1.0) snapshot = supervisor.snapshot() assert poll_calls >= 2 assert snapshot.host_path.available is True assert snapshot.endpoint.tcp_state == "reachable" await monitor.close() asyncio.run(scenario()) def test_verified_control_survives_thirty_seconds_of_alternating_unproven_association( ) -> None: async def scenario() -> None: monotonic_now = [100.0] suspend_aware_now = [1_000.0] supervisor = ConnectionSupervisor( clock=_clock, monotonic_clock=lambda: monotonic_now[0], suspend_aware_clock=lambda: suspend_aware_now[0], observation_ttl_seconds=2.0, ) supervisor.set_intent( intent_id="bridge-deferred-timeout", requested_mode="bridge", expected_device_id="k1-device-a", ) _apply_device_network(supervisor, intent_id="bridge-deferred-timeout") path = _available_path() epoch = supervisor.observe_host_path(path) assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-deferred-timeout", host_path_epoch=epoch, reachable=True, ) assert supervisor.observe_control_evidence( VerifiedControlEvidence( intent_id="bridge-deferred-timeout", transport_ref="ble-k1-a", host_path_epoch=epoch, target=TARGET, connection_mode="bridge", logical_device_id="k1-device-a", compatibility_profile_id="xgrids-k1-fw-3.0.2", control_session_id="control-deferred-timeout", ) ) initial = supervisor.snapshot() technical_timeout = HostPathProbeResult( available=False, fingerprint=None, interface=path.interface, source_ipv4=path.source_ipv4, route_class="unavailable", reason_code="host-wifi-operation-timeout", observation_failure_class="association-observer", kernel_route_fingerprint=path.kernel_route_fingerprint, ) privacy_limited = HostPathProbeResult( available=True, fingerprint="privacy-fallback-must-not-replace-proven-identity", interface=path.interface, source_ipv4=path.source_ipv4, route_class=path.route_class, reason_code="association-identity-unavailable", observation_failure_class="association-observer", kernel_route_fingerprint=path.kernel_route_fingerprint, ) host_calls = 0 tcp_calls = 0 async def host_probe(_target: EndpointTarget) -> HostPathProbeResult: nonlocal host_calls host_calls += 1 # Exercise both orderings inside a complete route/TCP/recheck # pass: timeout -> privacy and privacy -> timeout. return (technical_timeout, privacy_limited, privacy_limited, technical_timeout)[ (host_calls - 1) % 4 ] async def tcp_probe(_target: EndpointTarget) -> bool: nonlocal tcp_calls tcp_calls += 1 return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) refreshed = initial for _ in range(31): monotonic_now[0] += 1.0 suspend_aware_now[0] += 1.0 refreshed = await monitor.poll_once() assert refreshed.host_path.available is True assert refreshed.host_path.epoch == initial.host_path.epoch assert refreshed.host_path.fingerprint == initial.host_path.fingerprint assert refreshed.endpoint.tcp_state == "reachable" assert refreshed.endpoint.host_path_epoch == initial.host_path.epoch assert refreshed.device_identity.state == "verified" assert refreshed.control_plane.state == "healthy" assert refreshed.lease.state == "reachable" assert refreshed.lease.generation == initial.lease.generation assert refreshed.authority.control_allowed is True assert host_calls == 62 assert tcp_calls == 31 assert refreshed.revision == initial.revision + 62 # Retention refreshes only evidence actually sampled by the monitor. # If the monitor itself goes silent, the ordinary transport TTL still # fails closed instead of turning this continuity bridge into a lease. monotonic_now[0] += 2.01 suspend_aware_now[0] += 2.01 expired = supervisor.snapshot() assert expired.host_path.available is False assert expired.host_path.reason_code == "host-path-observation-stale" assert expired.authority.control_allowed is False assert expired.lease.state != "reachable" await monitor.close() asyncio.run(scenario()) def test_unproven_association_retention_still_revokes_on_exact_tcp_loss() -> None: async def scenario() -> None: supervisor, initial_epoch = _reachable_supervisor() initial = supervisor.snapshot() path = _available_path() technical_timeout = HostPathProbeResult( available=False, fingerprint=None, interface=path.interface, source_ipv4=path.source_ipv4, route_class="unavailable", reason_code="host-wifi-operation-timeout", observation_failure_class="association-observer", kernel_route_fingerprint=path.kernel_route_fingerprint, ) host_calls = 0 async def host_probe(_target: EndpointTarget) -> HostPathProbeResult: nonlocal host_calls host_calls += 1 return technical_timeout async def tcp_probe(_target: EndpointTarget) -> TcpReachabilityProbeResult: return TcpReachabilityProbeResult( reachable=False, reason_code="tcp-connect-refused", ) monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) lost = await monitor.poll_once() assert host_calls == 2 assert lost.host_path.available is True assert lost.host_path.epoch == initial_epoch assert lost.host_path.fingerprint == initial.host_path.fingerprint assert lost.endpoint.tcp_state == "unreachable" assert lost.endpoint.reason_code == "tcp-connect-refused" assert lost.control_plane.state == "lost" assert lost.control_plane.reason_code == "tcp-connect-refused" assert lost.lease.state != "reachable" assert lost.authority.control_allowed is False await monitor.close() asyncio.run(scenario()) def test_proven_association_change_after_bridged_tcp_rotates_epoch_immediately() -> None: async def scenario() -> None: supervisor, initial_epoch = _reachable_supervisor() initial = supervisor.snapshot() path = _available_path() technical_timeout = HostPathProbeResult( available=False, fingerprint=None, interface=path.interface, source_ipv4=path.source_ipv4, route_class="unavailable", reason_code="host-wifi-operation-timeout", observation_failure_class="association-observer", kernel_route_fingerprint=path.kernel_route_fingerprint, ) changed_association = HostPathProbeResult( available=True, fingerprint="same-kernel-route:new-proven-bssid-token", interface=path.interface, source_ipv4=path.source_ipv4, route_class=path.route_class, kernel_route_fingerprint=path.kernel_route_fingerprint, ) samples = iter((technical_timeout, changed_association)) async def host_probe(_target: EndpointTarget) -> HostPathProbeResult: return next(samples) async def tcp_probe(_target: EndpointTarget) -> bool: return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) changed = await monitor.poll_once() assert changed.host_path.available is True assert changed.host_path.epoch == initial_epoch + 1 assert changed.host_path.fingerprint == changed_association.fingerprint assert changed.host_path.kernel_route_fingerprint == ( initial.host_path.kernel_route_fingerprint ) assert changed.endpoint.tcp_state == "unknown" assert changed.device_identity.state == "stale" assert changed.control_plane.state == "lost" assert changed.lease.state != "reachable" assert changed.authority.control_allowed is False await monitor.close() asyncio.run(scenario()) def test_control_loss_during_unproven_recheck_cannot_be_restored_by_tcp_success() -> None: async def scenario() -> None: supervisor, initial_epoch = _reachable_supervisor() path = _available_path() technical_timeout = HostPathProbeResult( available=False, fingerprint=None, interface=path.interface, source_ipv4=path.source_ipv4, route_class="unavailable", reason_code="host-wifi-operation-timeout", observation_failure_class="association-observer", kernel_route_fingerprint=path.kernel_route_fingerprint, ) async def host_probe(_target: EndpointTarget) -> HostPathProbeResult: return technical_timeout async def tcp_probe(_target: EndpointTarget) -> bool: assert supervisor.observe_control_loss( intent_id="bridge-1", host_path_epoch=initial_epoch, control_session_id="control-session-a", reason_code="mqtt-control-session-closed", ) return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) lost = await monitor.poll_once() assert lost.host_path.epoch == initial_epoch assert lost.control_plane.state == "lost" assert lost.control_plane.reason_code == "mqtt-control-session-closed" assert lost.lease.state != "reachable" assert lost.authority.control_allowed is False await monitor.close() asyncio.run(scenario()) def test_control_proof_ttl_still_expires_during_healthy_transport_retention() -> None: async def scenario() -> None: monotonic_now = [100.0] suspend_aware_now = [1_000.0] supervisor = ConnectionSupervisor( clock=_clock, monotonic_clock=lambda: monotonic_now[0], suspend_aware_clock=lambda: suspend_aware_now[0], observation_ttl_seconds=2.0, control_proof_ttl_seconds=3.0, ) supervisor.set_intent( intent_id="bridge-control-ttl", requested_mode="bridge", expected_device_id="k1-device-a", ) _apply_device_network(supervisor, intent_id="bridge-control-ttl") path = _available_path() epoch = supervisor.observe_host_path(path) assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-control-ttl", host_path_epoch=epoch, reachable=True, ) assert supervisor.observe_control_evidence( VerifiedControlEvidence( intent_id="bridge-control-ttl", transport_ref="ble-k1-a", host_path_epoch=epoch, target=TARGET, connection_mode="bridge", logical_device_id="k1-device-a", compatibility_profile_id="xgrids-k1-fw-3.0.2", control_session_id="control-ttl-during-retention", ) ) technical_timeout = HostPathProbeResult( available=False, fingerprint=None, interface=path.interface, source_ipv4=path.source_ipv4, route_class="unavailable", reason_code="host-wifi-operation-timeout", observation_failure_class="association-observer", kernel_route_fingerprint=path.kernel_route_fingerprint, ) async def host_probe(_target: EndpointTarget) -> HostPathProbeResult: return technical_timeout async def tcp_probe(_target: EndpointTarget) -> bool: return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) for _ in range(3): monotonic_now[0] += 0.9 suspend_aware_now[0] += 0.9 fresh = await monitor.poll_once() assert fresh.host_path.epoch == epoch assert fresh.endpoint.tcp_state == "reachable" assert fresh.authority.control_allowed is True monotonic_now[0] += 0.31 suspend_aware_now[0] += 0.31 expired = await monitor.poll_once() assert expired.control_plane.state == "lost" assert expired.control_plane.reason_code == "control-proof-observation-stale" assert expired.lease.state != "reachable" assert expired.authority.control_allowed is False await monitor.close() asyncio.run(scenario()) def test_same_path_positive_reducer_observation_resets_technical_failure_streak() -> None: async def scenario() -> None: supervisor, initial_epoch = _configured_unverified_supervisor() path = _available_path() technical_timeout = HostPathProbeResult( available=False, fingerprint=None, interface=path.interface, source_ipv4=path.source_ipv4, route_class="unavailable", reason_code="host-wifi-operation-timeout", observation_failure_class="association-observer", kernel_route_fingerprint=path.kernel_route_fingerprint, ) async def host_probe(_target: EndpointTarget) -> HostPathProbeResult: return technical_timeout async def tcp_probe(_target: EndpointTarget) -> bool: raise AssertionError("technical failure must skip TCP") monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) first = await monitor.poll_once() second = await monitor.poll_once() assert first.authority.control_allowed is False assert second.authority.control_allowed is False refreshed_epoch = supervisor.observe_host_path(path) assert refreshed_epoch == initial_epoch external_positive = supervisor.snapshot() after_reset_first = await monitor.poll_once() after_reset_second = await monitor.poll_once() for deferred in (after_reset_first, after_reset_second): assert deferred.revision == external_positive.revision assert deferred.host_path == external_positive.host_path assert deferred.authority.control_allowed is False confirmed = await monitor.poll_once() assert confirmed.host_path.available is False assert confirmed.host_path.reason_code == "host-wifi-operation-timeout" assert confirmed.authority.control_allowed is False await monitor.close() asyncio.run(scenario()) def test_unrelated_supervisor_revisions_preserve_exact_configured_route() -> None: async def scenario() -> None: supervisor, epoch = _configured_unverified_supervisor() path = _available_path() initial = supervisor.snapshot() technical_timeout = HostPathProbeResult( available=False, fingerprint=None, interface=path.interface, source_ipv4=path.source_ipv4, route_class="unavailable", reason_code="host-wifi-operation-timeout", observation_failure_class="association-observer", kernel_route_fingerprint=path.kernel_route_fingerprint, ) async def host_probe(_target: EndpointTarget) -> HostPathProbeResult: return technical_timeout tcp_calls = 0 async def tcp_probe(_target: EndpointTarget) -> bool: nonlocal tcp_calls tcp_calls += 1 return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) first = await monitor.poll_once() assert first.host_path.available is True assert first.host_path.epoch == initial.host_path.epoch assert first.endpoint.tcp_state == "reachable" assert first.authority.control_allowed is False assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-1", host_path_epoch=epoch, reachable=True, ) endpoint_refresh = supervisor.snapshot() assert endpoint_refresh.revision > first.revision assert endpoint_refresh.host_path.epoch == initial.host_path.epoch assert endpoint_refresh.host_path.fingerprint == initial.host_path.fingerprint second = await monitor.poll_once() assert second.host_path.available is True assert second.host_path.epoch == initial.host_path.epoch assert second.endpoint.tcp_state == "reachable" assert second.authority.control_allowed is False assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-1", host_path_epoch=epoch, reachable=True, ) confirmed = await monitor.poll_once() assert confirmed.host_path.available is True assert confirmed.host_path.epoch == initial.host_path.epoch assert confirmed.endpoint.tcp_state == "reachable" assert confirmed.authority.control_allowed is False assert tcp_calls == 3 await monitor.close() asyncio.run(scenario()) def test_inflight_timeout_cannot_overwrite_a_concurrent_positive_observation() -> None: async def scenario() -> None: supervisor, initial_epoch = _configured_unverified_supervisor() path = _available_path() technical_timeout = HostPathProbeResult( available=False, fingerprint=None, interface=path.interface, source_ipv4=path.source_ipv4, route_class="unavailable", reason_code="host-wifi-operation-timeout", observation_failure_class="association-observer", kernel_route_fingerprint=path.kernel_route_fingerprint, ) timeout_probe_entered = asyncio.Event() release_timeout_probe = asyncio.Event() host_calls = 0 async def host_probe(_target: EndpointTarget) -> HostPathProbeResult: nonlocal host_calls host_calls += 1 if host_calls == 1: timeout_probe_entered.set() await release_timeout_probe.wait() return technical_timeout async def tcp_probe(_target: EndpointTarget) -> bool: return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) inflight_timeout = asyncio.create_task(monitor.poll_once()) await timeout_probe_entered.wait() refreshed_epoch = supervisor.observe_host_path(path) assert refreshed_epoch == initial_epoch external_positive = supervisor.snapshot() release_timeout_probe.set() raced_timeout = await inflight_timeout assert raced_timeout.revision >= external_positive.revision assert raced_timeout.host_path.available is True assert raced_timeout.host_path.epoch == external_positive.host_path.epoch assert raced_timeout.host_path.fingerprint == external_positive.host_path.fingerprint assert raced_timeout.endpoint.tcp_state == "reachable" assert raced_timeout.authority.control_allowed is False await monitor.close() asyncio.run(scenario()) def test_observation_ttl_revokes_authority_after_monitor_silence() -> None: monotonic_now = [100.0] supervisor = ConnectionSupervisor( clock=_clock, monotonic_clock=lambda: monotonic_now[0], observation_ttl_seconds=2.0, ) supervisor.set_intent( intent_id="bridge-ttl", requested_mode="bridge", expected_device_id="k1-device-a", ) _apply_device_network(supervisor, intent_id="bridge-ttl") epoch = supervisor.observe_host_path(_available_path()) assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-ttl", host_path_epoch=epoch, reachable=True, ) assert supervisor.observe_control_evidence( VerifiedControlEvidence( intent_id="bridge-ttl", transport_ref="ble-k1-a", host_path_epoch=epoch, target=TARGET, connection_mode="bridge", logical_device_id="k1-device-a", compatibility_profile_id="xgrids-k1-fw-3.0.2", control_session_id="control-ttl", ) ) assert supervisor.observe_data_plane( intent_id="bridge-ttl", host_path_epoch=epoch, state="healthy", session_id="data-session-route-ttl", ) assert supervisor.snapshot().authority.control_allowed is True monotonic_now[0] += 2.01 expired = supervisor.snapshot() assert expired.host_path.epoch == epoch + 1 assert expired.host_path.available is False assert expired.host_path.reason_code == "host-path-observation-stale" assert expired.endpoint.tcp_state == "unknown" assert expired.device_identity.state == "stale" assert expired.control_plane.state == "lost" assert expired.lease.state == "configured-unverified" assert expired.authority.control_allowed is False assert expired.authority.data_ingest_authoritative is False assert expired.data_plane.state == "healthy" assert expired.data_plane.session_id == "data-session-route-ttl" assert expired.data_plane.host_path_epoch == epoch assert expired.last_known is not None def test_endpoint_ttl_revokes_authority_even_when_route_observation_stays_fresh() -> None: monotonic_now = [200.0] supervisor = ConnectionSupervisor( clock=_clock, monotonic_clock=lambda: monotonic_now[0], observation_ttl_seconds=2.0, ) supervisor.set_intent( intent_id="quick-ttl", requested_mode="quick-connect", expected_device_id="k1-device-a", ) _apply_device_network( supervisor, intent_id="quick-ttl", connection_mode="quick-connect", ) epoch = supervisor.observe_host_path(_available_path()) assert supervisor.observe_endpoint( target=TARGET, intent_id="quick-ttl", host_path_epoch=epoch, reachable=True, ) assert supervisor.observe_control_evidence( VerifiedControlEvidence( intent_id="quick-ttl", transport_ref="ble-k1-a", host_path_epoch=epoch, target=TARGET, connection_mode="quick-connect", logical_device_id="k1-device-a", compatibility_profile_id="xgrids-k1-fw-3.0.2", control_session_id="quick-control-ttl", ) ) assert supervisor.observe_data_plane( intent_id="quick-ttl", host_path_epoch=epoch, state="healthy", session_id="data-session-endpoint-ttl", ) monotonic_now[0] += 1.5 assert supervisor.observe_host_path(_available_path()) == epoch monotonic_now[0] += 0.6 expired = supervisor.snapshot() assert expired.host_path.available is True assert expired.host_path.epoch == epoch assert expired.endpoint.tcp_state == "unreachable" assert expired.endpoint.reason_code == "endpoint-observation-stale" assert expired.control_plane.state == "lost" assert expired.lease.state == "configured-unverified" assert expired.authority.control_allowed is False assert expired.authority.data_ingest_authoritative is False assert expired.data_plane.state == "healthy" assert expired.data_plane.session_id == "data-session-endpoint-ttl" assert expired.data_plane.host_path_epoch == epoch def test_suspend_elapsed_time_revokes_authority_when_macos_monotonic_stops() -> None: monotonic_now = [300.0] suspend_aware_now = [1_000.0] supervisor = ConnectionSupervisor( clock=_clock, monotonic_clock=lambda: monotonic_now[0], suspend_aware_clock=lambda: suspend_aware_now[0], observation_ttl_seconds=2.0, ) supervisor.set_intent( intent_id="bridge-suspend", requested_mode="bridge", expected_device_id="k1-device-a", ) _apply_device_network(supervisor, intent_id="bridge-suspend") epoch = supervisor.observe_host_path(_available_path()) assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-suspend", host_path_epoch=epoch, reachable=True, ) assert supervisor.observe_control_evidence( VerifiedControlEvidence( intent_id="bridge-suspend", transport_ref="ble-k1-a", host_path_epoch=epoch, target=TARGET, connection_mode="bridge", logical_device_id="k1-device-a", compatibility_profile_id="xgrids-k1-fw-3.0.2", control_session_id="control-before-sleep", ) ) # mach_absolute_time/time.monotonic may remain unchanged while macOS is # asleep. Wall elapsed time must still invalidate every pre-sleep proof. suspend_aware_now[0] += 60.0 expired = supervisor.snapshot() assert monotonic_now[0] == 300.0 assert expired.host_path.epoch == epoch + 1 assert expired.host_path.reason_code == "host-path-observation-stale" assert expired.authority.control_allowed is False assert expired.authority.acquisition_start_allowed is False def test_control_proof_ttl_is_independent_from_fresh_route_and_tcp_observations() -> None: monotonic_now = [400.0] suspend_aware_now = [4_000.0] supervisor = ConnectionSupervisor( clock=_clock, monotonic_clock=lambda: monotonic_now[0], suspend_aware_clock=lambda: suspend_aware_now[0], observation_ttl_seconds=300.0, control_proof_ttl_seconds=2.0, ) supervisor.set_intent( intent_id="bridge-control-proof-ttl", requested_mode="bridge", expected_device_id="k1-device-a", ) _apply_device_network(supervisor, intent_id="bridge-control-proof-ttl") epoch = supervisor.observe_host_path(_available_path()) assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-control-proof-ttl", host_path_epoch=epoch, reachable=True, ) assert supervisor.observe_control_evidence( VerifiedControlEvidence( intent_id="bridge-control-proof-ttl", transport_ref="ble-k1-a", host_path_epoch=epoch, target=TARGET, connection_mode="bridge", logical_device_id="k1-device-a", compatibility_profile_id="xgrids-k1-fw-3.0.2", control_session_id="control-proof-session", control_proof_revision=4, control_proof_source="correlated-application-response", ) ) assert supervisor.observe_data_plane( intent_id="bridge-control-proof-ttl", host_path_epoch=epoch, state="healthy", session_id="data-session-generation-10", ) # Replaying the same API snapshot is idempotent; it is not new remote # evidence and therefore cannot extend authority. monotonic_now[0] += 1.5 suspend_aware_now[0] += 1.5 assert ( supervisor.refresh_control_evidence( intent_id="bridge-control-proof-ttl", host_path_epoch=epoch, control_session_id="control-proof-session", control_proof_revision=4, control_proof_source="correlated-application-response", ) is False ) assert supervisor.observe_host_path(_available_path()) == epoch assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-control-proof-ttl", host_path_epoch=epoch, reachable=True, ) monotonic_now[0] += 0.6 suspend_aware_now[0] += 0.6 expired = supervisor.snapshot() assert expired.host_path.available is True assert expired.endpoint.tcp_state == "reachable" assert expired.device_identity.state == "stale" assert expired.control_plane.state == "lost" assert expired.control_plane.reason_code == "control-proof-observation-stale" assert expired.lease.state == "configured-unverified" assert expired.authority.control_allowed is False assert expired.authority.acquisition_start_allowed is False assert expired.authority.data_ingest_authoritative is False assert expired.data_plane.state == "healthy" assert expired.data_plane.session_id == "data-session-generation-10" assert expired.data_plane.host_path_epoch == epoch assert [item.code for item in expired.diagnostics] == ["host.mqtt.transport-unavailable"] assert supervisor.observe_data_plane( intent_id="bridge-control-proof-ttl", host_path_epoch=epoch, state="healthy", session_id="data-session-generation-10", ) evidence_only = supervisor.snapshot() assert evidence_only.control_plane.state == "lost" assert evidence_only.data_plane.state == "healthy" assert evidence_only.authority.control_allowed is False assert evidence_only.authority.data_ingest_authoritative is False def test_new_exact_control_proof_revision_refreshes_dual_clock_authority() -> None: monotonic_now = [500.0] suspend_aware_now = [5_000.0] supervisor = ConnectionSupervisor( clock=_clock, monotonic_clock=lambda: monotonic_now[0], suspend_aware_clock=lambda: suspend_aware_now[0], observation_ttl_seconds=300.0, control_proof_ttl_seconds=2.0, ) supervisor.set_intent( intent_id="bridge-control-proof-refresh", requested_mode="bridge", expected_device_id="k1-device-a", ) _apply_device_network(supervisor, intent_id="bridge-control-proof-refresh") epoch = supervisor.observe_host_path(_available_path()) assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-control-proof-refresh", host_path_epoch=epoch, reachable=True, ) assert supervisor.observe_control_evidence( VerifiedControlEvidence( intent_id="bridge-control-proof-refresh", transport_ref="ble-k1-a", host_path_epoch=epoch, target=TARGET, connection_mode="bridge", logical_device_id="k1-device-a", compatibility_profile_id="xgrids-k1-fw-3.0.2", control_session_id="control-proof-refresh-session", control_proof_revision=1, ) ) monotonic_now[0] += 1.8 suspend_aware_now[0] += 1.8 assert supervisor.refresh_control_evidence( intent_id="bridge-control-proof-refresh", host_path_epoch=epoch, control_session_id="control-proof-refresh-session", control_proof_revision=2, control_proof_source="mqtt-heartbeat", ) monotonic_now[0] += 1.8 suspend_aware_now[0] += 1.8 still_fresh = supervisor.snapshot() assert still_fresh.authority.control_allowed is True # macOS monotonic can stop while asleep; wall elapsed time alone must # revoke the refreshed proof without waiting for a route transition. suspend_aware_now[0] += 60.0 expired = supervisor.snapshot() assert expired.host_path.available is True assert expired.endpoint.tcp_state == "reachable" assert expired.control_plane.reason_code == "control-proof-observation-stale" assert expired.authority.control_allowed is False def test_data_error_before_first_packet_is_an_ignored_observation() -> None: supervisor, epoch = _reachable_supervisor() assert ( supervisor.observe_data_plane( intent_id="bridge-1", host_path_epoch=epoch, state="lost", session_id=None, reason_code="live-runtime-error", ) is False ) snapshot = supervisor.snapshot() assert snapshot.data_plane.state == "idle" assert snapshot.authority.control_allowed is True def test_monitor_rechecks_route_after_tcp_and_rejects_success_from_old_wifi() -> None: async def scenario() -> None: supervisor = ConnectionSupervisor(clock=_clock) supervisor.set_intent(intent_id="bridge-race", requested_mode="bridge") _apply_device_network(supervisor, intent_id="bridge-race") paths = iter( ( _available_path("en0:192.168.68.20:router-a"), _available_path("en0:192.168.77.20:router-b"), ) ) async def host_probe(target: EndpointTarget) -> HostPathProbeResult: assert target == TARGET return next(paths) async def tcp_probe(target: EndpointTarget) -> bool: assert target == TARGET return True monitor = ReadOnlyConnectionMonitor( supervisor, host_path_probe=host_probe, tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) snapshot = await monitor.poll_once() assert snapshot.host_path.fingerprint == "en0:192.168.77.20:router-b" assert snapshot.host_path.epoch == 2 assert snapshot.endpoint.tcp_state == "unknown" assert snapshot.authority.control_allowed is False await monitor.close() asyncio.run(scenario())