NODEDC_MISSION_CORE/tests/test_xgrids_application_session.py

from __future__ import annotations

import hashlib
import logging
import threading
import time
from collections.abc import Callable, Collection, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import pytest

from k1link.device_plugins.xgrids_k1.physical_command_coordinator import (
    LedgerPhysicalCommandCoordinator,
    PhysicalCommandIntentContext,
    PhysicalCommandRuntimeBinding,
)
from k1link.device_plugins.xgrids_k1.physical_command_ledger import (
    PhysicalCommandConnectionBinding,
    PhysicalCommandIdentity,
    PhysicalCommandLedger,
    PhysicalCommandStatusEvidence,
)
from k1link.device_plugins.xgrids_k1.protocol import application_session as session_module
from k1link.device_plugins.xgrids_k1.protocol.application_acceptance import (
    ApplicationAcceptanceError,
)
from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import (
    COMPATIBILITY_PROFILE_ID,
    ApplicationControlAuthority,
    LiveDeviceControlBinding,
)
from k1link.device_plugins.xgrids_k1.protocol.application_mqtt import (
    ApplicationControlProofStale,
    ApplicationMqttTransportError,
)
from k1link.device_plugins.xgrids_k1.protocol.application_publish import (
    OneShotPublishEnvelope,
)
from k1link.device_plugins.xgrids_k1.protocol.application_session import (
    ApplicationConnectionBinding,
    ApplicationConnectionBindingLost,
    ApplicationControlProofExpired,
    ApplicationControlStateConflict,
    ApplicationStartCheckpointSettlementError,
    InteractiveApplicationControlSession,
    OperatorPresenceConfirmation,
)
from k1link.device_plugins.xgrids_k1.protocol.modeling_control import OPENAPI_SUCCESS

APPLICATION_KEY = "00000000-0000-0000-0000-000000000000"
RETIRED_VENDOR_DEVICE_ID = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
RETIRED_DEVICE_SERIAL = "K1SERIAL01"


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


def _test_uint(number: int, value: int) -> bytes:
    return _test_varint(number << 3) + _test_varint(value)


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


def _test_text(number: int, value: str) -> bytes:
    return _test_bytes(number, value.encode())


def _test_response_header(session_id: str) -> bytes:
    return b"".join(
        (
            _test_text(4, RETIRED_VENDOR_DEVICE_ID),
            _test_text(5, session_id),
            _test_text(6, APPLICATION_KEY),
        )
    )


def _test_device_info_response(session_id: str) -> bytes:
    base_info = b"".join(
        (
            _test_text(2, "V3.0.2_20250624.122658"),
            _test_text(3, "V3.0.2"),
            _test_text(6, "LixelKity K1"),
            _test_text(7, RETIRED_DEVICE_SERIAL),
            _test_text(8, "A4"),
        )
    )
    device_info = _test_bytes(2, base_info) + _test_bytes(7, _test_uint(1, 1))
    return b"".join(
        (
            _test_bytes(1, _test_response_header(session_id)),
            _test_bytes(2, device_info),
            _test_bytes(15, _test_uint(1, OPENAPI_SUCCESS)),
        )
    )


def _test_generic_response(session_id: str) -> bytes:
    return _test_bytes(1, _test_response_header(session_id)) + _test_bytes(
        15,
        _test_uint(1, OPENAPI_SUCCESS),
    )


class FakeAuthorityLoader:
    def __init__(self) -> None:
        self.calls = 0

    def load(self) -> ApplicationControlAuthority:
        self.calls += 1
        return ApplicationControlAuthority(openapi_key=APPLICATION_KEY)


@dataclass
class FakeTransportSnapshot:
    state: str
    ready: bool
    proof_fresh: bool
    publish_attempts: int = 0

    def as_dict(self) -> dict[str, object]:
        return {
            "state": self.state,
            "publish_attempts": self.publish_attempts,
            "qos2_completions": 0,
            "correlated_responses": 0,
            "ignored_known_responses": 0,
            "late_known_responses": 0,
            "latest_device_session_state": "ready" if self.ready else "scanning",
            "latest_device_project_bound": not self.ready,
            "automatic_retry": False,
            "automatic_reconnect": False,
            "control_proof_revision": 1,
            "control_proof_source": "correlated-application-response",
            "control_proof_fresh": self.proof_fresh,
            "control_proof_age_seconds": 0.0,
        }


class FakeTransport:
    def __init__(self, host: str) -> None:
        self.host = host
        self.state = "new"
        self.ready = True
        self.proof_fresh = True
        self.publish_attempts = 0
        self.evidence_observer: object | None = None
        self.dispatch_guard: Callable[[], None] | None = None

    def install_evidence_observer(self, observer: object) -> None:
        assert self.state == "new"
        self.evidence_observer = observer

    def install_dispatch_guard(self, guard: Callable[[], None]) -> None:
        assert self.state == "new"
        self.dispatch_guard = guard

    def open(self) -> FakeTransportSnapshot:
        self.state = "ready"
        return self.snapshot()

    def close(self) -> None:
        self.state = "closed"

    def snapshot(self) -> FakeTransportSnapshot:
        return FakeTransportSnapshot(
            self.state,
            self.ready,
            self.proof_fresh,
            self.publish_attempts,
        )

    def validate_control_proof(self, _binding: LiveDeviceControlBinding) -> None:
        if not self.proof_fresh:
            raise ApplicationControlProofStale("test control proof expired")


class FakePhysicalCommandCoordinator:
    def __init__(self, *, fail_prepare: bool = False) -> None:
        self.fail_prepare = fail_prepare
        self.bindings: list[PhysicalCommandRuntimeBinding] = []
        self.prepares: list[tuple[PhysicalCommandIntentContext, str, object]] = []
        self.resolutions: list[tuple[str, str | None]] = []
        self.phase_probe: Callable[[], str] | None = None
        self.snapshot_override: dict[str, object] | None = None
        self.snapshot_calls = 0

    def bind_control_session(self, binding: PhysicalCommandRuntimeBinding) -> None:
        self.bindings.append(binding)

    def prepare(
        self,
        context: PhysicalCommandIntentContext,
        *,
        action: str,
        envelope: object,
    ) -> None:
        self.prepares.append((context, action, envelope))
        if self.fail_prepare:
            raise RuntimeError("durable prepare failed")

    def resolve(self, action: str) -> None:
        phase = self.phase_probe() if self.phase_probe is not None else None
        self.resolutions.append((action, phase))

    def resolve_prepared_not_dispatched(self, action: str) -> None:
        phase = self.phase_probe() if self.phase_probe is not None else None
        self.resolutions.append((f"{action}-not-dispatched", phase))

    def snapshot(self) -> dict[str, object]:
        self.snapshot_calls += 1
        if self.snapshot_override is not None:
            return self.snapshot_override
        return {
            "status": "test",
            "requires_reconciliation": False,
            "automatic_replay_allowed": False,
        }

    def publish_dispatching(
        self,
        _evidence: object,
        *,
        dispatch_admission_deadline_reached: Callable[[], bool] | None = None,
        dispatch_admission_commit: Callable[[], None] | None = None,
    ) -> None:
        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()
        return

    def publish_result(
        self,
        _evidence: object,
        *,
        publish_call_returned: bool,
    ) -> None:
        del publish_call_returned

    def qos2_completed(self, _evidence: object) -> None:
        return

    def application_response(self, _evidence: object) -> None:
        return

    def device_status(self, _evidence: object) -> None:
        return


class FakeExecutor:
    records: list[str] = []

    def __init__(self, transport: FakeTransport) -> None:
        self.transport = transport
        self.binding = LiveDeviceControlBinding(
            vendor_device_id="device-id",
            device_serial="serial-id",
            software_version="3.0.2",
            system_version="3.0.2",
            device_model="LixelKity K1",
            device_type="A4",
            is_activated=True,
        )

    def run_connection_stage(self, _orchestrator: object) -> LiveDeviceControlBinding:
        self.records.append("connection:1-6")
        return self.binding

    def run_read_only_inspection_stage(
        self,
        _orchestrator: object,
    ) -> LiveDeviceControlBinding:
        self.records.append("inspection:1")
        return self.binding

    def complete_connection_stage(
        self,
        _orchestrator: object,
        *,
        expected_binding: LiveDeviceControlBinding,
    ) -> LiveDeviceControlBinding:
        assert expected_binding == self.binding
        self.records.append("connection:2-6")
        return self.binding

    def wait_for_operator_checkpoint(
        self,
        event: str,
        observed: Any,
        *,
        reconciled_active_observed: Any | None = None,
    ) -> str | None:
        self.records.append(f"wait:{event}")
        deadline = time.monotonic() + 2.0
        while not observed():
            if (
                reconciled_active_observed is not None
                and reconciled_active_observed()
            ):
                return None
            if time.monotonic() >= deadline:
                raise TimeoutError(f"test did not release {event}")
            threading.Event().wait(0.005)
        return event

    def adopt_reconciled_scanning(
        self,
        *,
        authority: ApplicationControlAuthority,
        binding: LiveDeviceControlBinding,
    ) -> None:
        del authority
        assert binding == self.binding
        self.records.append("adopt:scanning")

    def run_workspace_entry_stage(
        self,
        _orchestrator: object,
        _checkpoint: object,
        *,
        dispatch_guard: Callable[[], None] | None = None,
    ) -> LiveDeviceControlBinding:
        if dispatch_guard is not None:
            dispatch_guard()
        self.records.append("workspace:7")
        return self.binding

    def run_project_prompt_stage(
        self,
        _orchestrator: object,
        _checkpoint: object,
        *,
        dispatch_guard: Callable[[], None] | None = None,
    ) -> LiveDeviceControlBinding:
        if dispatch_guard is not None:
            dispatch_guard()
        self.records.append("project:8-10")
        return self.binding

    def execute_canonical_start(self, *_args: object, **_kwargs: object) -> object:
        self.records.append("start:11-14")
        self.transport.ready = False
        return object()

    def maintain_active_until_stop_requested(self, observed: Any) -> None:
        self.records.append("wait:stop")
        while not observed():
            threading.Event().wait(0.005)

    def execute_canonical_stop(self, *_args: object, **_kwargs: object) -> object:
        self.records.append("stop")
        self.transport.ready = True
        return object()

    def maintain_post_stop_until_standby(self) -> None:
        self.records.append("wait:device-standby")

    def snapshot(self) -> dict[str, object]:
        return {
            "records": list(self.records),
            "dialogue_stage": "test-stage",
            "start_attempted": False,
            "stop_attempted": False,
            "automatic_retry": False,
        }


def _confirmation() -> OperatorPresenceConfirmation:
    return OperatorPresenceConfirmation(
        operator_present=True,
        owner_controlled_device=True,
        lixelgo_closed=True,
        battery_storage_confirmed=True,
        expected_physical_state_confirmed=True,
    )


def _wait_phase(
    session: InteractiveApplicationControlSession,
    expected: str,
) -> dict[str, object]:
    # FakeExecutor may use its full two-second checkpoint timeout before the
    # worker publishes the terminal state. Keep the observer deadline strictly
    # larger so this helper does not race the transition it is asserting.
    deadline = time.monotonic() + 3.0
    while time.monotonic() < deadline:
        snapshot = session.snapshot()
        if snapshot["state"] == expected:
            return snapshot
        threading.Event().wait(0.005)
    raise AssertionError(f"session did not reach {expected}: {session.snapshot()}")


def _connection_binding() -> ApplicationConnectionBinding:
    return ApplicationConnectionBinding(
        intent_id="test-intent-1",
        transport_ref="test-transport-1",
        host_path_epoch=1,
        target_ipv4="192.168.1.20",
        target_port=1883,
        connection_mode="bridge",
    )


def _physical_context(action: str) -> PhysicalCommandIntentContext:
    return PhysicalCommandIntentContext(
        operation_id=f"operation-{action}-1",
        parent_operation_id=("operation-start-1" if action == "stop" else None),
        acquisition_id="acquisition-1",
    )


def _successful_start_checkpoint_observer(
    _phase: str,
    _context: PhysicalCommandIntentContext,
    _envelope: object,
) -> None:
    return None


def test_control_only_snapshot_reuses_preverified_physical_command() -> None:
    coordinator = FakePhysicalCommandCoordinator()
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=FakeTransport,  # type: ignore[arg-type]
        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
    )

    control_only = session.snapshot_control_only()
    assert control_only["state"] == "idle"
    assert control_only["physical_command"] is None
    assert coordinator.snapshot_calls == 0

    verified_physical = {
        "status": "resolved",
        "requires_reconciliation": False,
    }
    joined = session.snapshot_with_physical_command(verified_physical)
    assert joined["physical_command"] == verified_physical
    assert coordinator.snapshot_calls == 0

    regular = session.snapshot()
    assert regular["physical_command"] == {
        "status": "test",
        "requires_reconciliation": False,
        "automatic_replay_allowed": False,
    }
    assert coordinator.snapshot_calls == 1


def test_canonical_stages_require_operator_events_but_device_standby_does_not(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    FakeExecutor.records = []
    scanning_observed = threading.Event()
    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        FakeExecutor,
    )
    loader = FakeAuthorityLoader()
    transport = FakeTransport("192.168.1.20")
    session = InteractiveApplicationControlSession(
        loader,
        transport_factory=lambda _host: transport,  # type: ignore[arg-type]
        epoch_seconds=lambda: 1_752_680_000,
        scanning_observer=scanning_observed.set,
    )

    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
        connection_binding=_connection_binding(),
    )
    connected = _wait_phase(session, "connection-ready")
    assert connected["verified_control"]["transport_ref"] == "test-transport-1"  # type: ignore[index]
    threading.Event().wait(0.02)
    assert FakeExecutor.records == [
        "inspection:1",
        "connection:2-6",
        "wait:workspace-entered",
    ]

    session.enter_workspace()
    _wait_phase(session, "workspace-ready")
    assert FakeExecutor.records[-1] == "wait:project-prompt-opened"

    session.open_project_prompt()
    _wait_phase(session, "project-ready")
    assert FakeExecutor.records[-1] == "wait:start-confirmed"

    session.request_start(project_name="TEST001", confirmation=_confirmation())
    _wait_phase(session, "scanning")
    assert scanning_observed.wait(timeout=1.0)
    assert FakeExecutor.records[-1] == "wait:stop"

    session.request_stop(confirmation=_confirmation())
    completed = _wait_phase(session, "completed")
    assert completed["can_confirm_standby"] is False
    assert completed["pending_operator_action"] is None
    assert loader.calls == 1
    assert completed["automatic_retry"] is False
    assert completed["scanning_observer_errors"] == 0
    assert completed["scripted_transitions"] is False
    assert FakeExecutor.records == [
        "inspection:1",
        "connection:2-6",
        "wait:workspace-entered",
        "workspace:7",
        "wait:project-prompt-opened",
        "project:8-10",
        "wait:start-confirmed",
        "start:11-14",
        "wait:stop",
        "stop",
        "wait:device-standby",
    ]


def test_session_checkpoints_do_not_repeat_native_dispatch_proof(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Slow host-path proof belongs to the exact publish lease, not UI checkpoints."""

    class DispatchingExecutor(FakeExecutor):
        def _dispatch_one_packet(self) -> None:
            guard = self.transport.dispatch_guard
            assert guard is not None
            release = guard()
            try:
                self.transport.publish_attempts += 1
            finally:
                if callable(release):
                    release()

        def run_workspace_entry_stage(
            self,
            orchestrator: object,
            checkpoint: object,
            *,
            dispatch_guard: Callable[[], None] | None = None,
        ) -> LiveDeviceControlBinding:
            if dispatch_guard is not None:
                dispatch_guard()
            self._dispatch_one_packet()
            return super().run_workspace_entry_stage(
                orchestrator,
                checkpoint,
                dispatch_guard=None,
            )

        def run_project_prompt_stage(
            self,
            orchestrator: object,
            checkpoint: object,
            *,
            dispatch_guard: Callable[[], None] | None = None,
        ) -> LiveDeviceControlBinding:
            if dispatch_guard is not None:
                dispatch_guard()
            self._dispatch_one_packet()
            return super().run_project_prompt_stage(
                orchestrator,
                checkpoint,
                dispatch_guard=None,
            )

        def execute_canonical_start(self, *_args: object, **kwargs: object) -> object:
            dispatch_guard = kwargs.get("dispatch_guard")
            if callable(dispatch_guard):
                dispatch_guard()
            self._dispatch_one_packet()
            return super().execute_canonical_start(*_args, **kwargs)

        def execute_canonical_stop(self, *_args: object, **kwargs: object) -> object:
            dispatch_guard = kwargs.get("dispatch_guard")
            if callable(dispatch_guard):
                dispatch_guard()
            self._dispatch_one_packet()
            return super().execute_canonical_stop(*_args, **kwargs)

    FakeExecutor.records = []
    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        DispatchingExecutor,
    )
    initial_path_proofs: list[ApplicationConnectionBinding] = []
    full_binding_proofs: list[ApplicationConnectionBinding] = []
    snapshot_proofs: list[ApplicationConnectionBinding] = []
    dispatch_proofs: list[ApplicationConnectionBinding] = []
    dispatch_releases: list[ApplicationConnectionBinding] = []

    def acquire_dispatch_proof(
        binding: ApplicationConnectionBinding,
        _deadline_reached: Callable[[], bool] | None,
    ) -> Callable[[], None]:
        dispatch_proofs.append(binding)
        return lambda: dispatch_releases.append(binding)

    transport = FakeTransport("192.168.1.20")
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda _host: transport,  # type: ignore[arg-type]
        connection_path_validator=lambda binding: initial_path_proofs.append(binding),
        connection_binding_validator=lambda binding: full_binding_proofs.append(binding),
        connection_binding_snapshot_validator=lambda binding: snapshot_proofs.append(binding),
        connection_dispatch_lease=acquire_dispatch_proof,
    )
    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
        connection_binding=_connection_binding(),
    )
    _wait_phase(session, "connection-ready")
    session.enter_workspace()
    _wait_phase(session, "workspace-ready")
    session.open_project_prompt()
    _wait_phase(session, "project-ready")
    session.request_start(project_name="TEST001", confirmation=_confirmation())
    _wait_phase(session, "scanning")
    session.request_stop(confirmation=_confirmation())
    _wait_phase(session, "completed")

    assert initial_path_proofs == [_connection_binding()]
    assert full_binding_proofs == []
    assert len(snapshot_proofs) >= 12
    assert dispatch_proofs == [_connection_binding()] * 4
    assert dispatch_releases == dispatch_proofs


def test_exact_publish_lease_still_rejects_epoch_drift_after_snapshot_check(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A cheap checkpoint pass can never bypass the final physical route fence."""

    class DispatchRejectedExecutor(FakeExecutor):
        def run_workspace_entry_stage(
            self,
            orchestrator: object,
            checkpoint: object,
            *,
            dispatch_guard: Callable[[], None] | None = None,
        ) -> LiveDeviceControlBinding:
            if dispatch_guard is not None:
                dispatch_guard()
            guard = self.transport.dispatch_guard
            assert guard is not None
            guard()
            return super().run_workspace_entry_stage(
                orchestrator,
                checkpoint,
                dispatch_guard=None,
            )

    FakeExecutor.records = []
    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        DispatchRejectedExecutor,
    )
    snapshot_proofs: list[ApplicationConnectionBinding] = []

    def reject_dispatch(
        _binding: ApplicationConnectionBinding,
        _deadline_reached: Callable[[], bool] | None,
    ) -> Callable[[], None]:
        raise ApplicationConnectionBindingLost("test host-path epoch changed")

    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
        connection_path_validator=lambda _binding: True,
        connection_binding_validator=lambda _binding: True,
        connection_binding_snapshot_validator=lambda binding: snapshot_proofs.append(binding),
        connection_dispatch_lease=reject_dispatch,
    )
    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
        connection_binding=_connection_binding(),
    )
    _wait_phase(session, "connection-ready")
    session.enter_workspace()
    failed = _wait_phase(session, "failed")

    assert snapshot_proofs
    assert "workspace:7" not in FakeExecutor.records
    assert failed["failure"]["reason_code"] == (  # type: ignore[index]
        ApplicationConnectionBindingLost.reason_code
    )


def test_read_only_device_info_open_does_not_require_physical_acceptance(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Connect may prove DeviceInfo without inventing START/STOP confirmations."""

    FakeExecutor.records = []
    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        FakeExecutor,
    )
    coordinator = FakePhysicalCommandCoordinator()
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
    )

    opened = session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        connection_binding=_connection_binding(),
    )
    assert opened["state"] in {"connecting", "connection-ready"}
    ready = _wait_phase(session, "connection-ready")

    assert ready["verified_control"] is not None
    assert FakeExecutor.records[:2] == ["inspection:1", "connection:2-6"]
    assert coordinator.prepares == []
    assert "start:11-14" not in FakeExecutor.records
    assert "stop" not in FakeExecutor.records
    session.close_prestart(
        expected_session_generation=ready["session_generation"],  # type: ignore[arg-type]
        expected_state_revision=ready["state_revision"],  # type: ignore[arg-type]
    )


def test_inspection_session_rejects_workspace_and_closes_without_connection_stage(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    FakeExecutor.records = []
    monkeypatch.setattr(session_module, "PhysicalAcceptanceDialogueExecutor", FakeExecutor)
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
    )

    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        connection_binding=_connection_binding(),
        inspection_only=True,
    )
    ready = _wait_phase(session, "connection-ready")

    with pytest.raises(ApplicationAcceptanceError, match="read-only inspection"):
        session.enter_workspace(
            expected_session_generation=ready["session_generation"],  # type: ignore[arg-type]
            expected_state_revision=ready["state_revision"],  # type: ignore[arg-type]
        )
    assert FakeExecutor.records == ["inspection:1", "wait:workspace-entered"]

    session.close_prestart(
        expected_session_generation=ready["session_generation"],  # type: ignore[arg-type]
        expected_state_revision=ready["state_revision"],  # type: ignore[arg-type]
    )
    _wait_phase(session, "closed")
    assert FakeExecutor.records == ["inspection:1", "wait:workspace-entered"]


@pytest.mark.parametrize(
    ("reconciliation_kind", "record_resolution"),
    [
        ("ambiguous-outcome", "physical-active-observed"),
        ("resolved-active-rebind", "start-active-observed"),
        ("prepared-stop-classification", "not-dispatched"),
    ],
)
def test_reconciled_scanning_adoption_emits_no_start_and_one_explicit_stop(
    monkeypatch: pytest.MonkeyPatch,
    reconciliation_kind: str,
    record_resolution: str,
) -> None:
    FakeExecutor.records = []
    monkeypatch.setattr(session_module, "PhysicalAcceptanceDialogueExecutor", FakeExecutor)
    coordinator = FakePhysicalCommandCoordinator()
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
    )

    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        connection_binding=_connection_binding(),
        inspection_only=True,
    )
    ready = _wait_phase(session, "connection-ready")
    verified = ready["verified_control"]
    assert isinstance(verified, dict)
    reconciliation_id = "verify-operation-1.physical"
    exact_connection = {
        field: verified[field]
        for field in (
            "intent_id",
            "transport_ref",
            "connection_mode",
            "target_ipv4",
            "target_port",
            "host_path_epoch",
            "control_session_id",
            "producer_generation",
        )
    }
    coordinator.snapshot_override = {
        "status": "resolved",
        "requires_reconciliation": False,
        "resolved_active_recovery_required": True,
        "record": {
            "resolution": record_resolution,
            "reconciliations": [{
                "reconciliation_id": reconciliation_id,
                "kind": reconciliation_kind,
                "resolution": "physical-active-observed",
                "verified_binding": {"connection": exact_connection},
                "observation": {
                    "source": "explicit-read-only-reconciliation",
                    "control_session_id": verified["control_session_id"],
                    "host_path_epoch": verified["host_path_epoch"],
                    "producer_generation": verified["producer_generation"],
                    "session_state": "scanning",
                    "project_bound": True,
                    "init_ready": True,
                    "mqtt_retained": False,
                },
            }],
        },
    }

    requested = session.adopt_reconciled_scanning(
        reconciliation_id=reconciliation_id,
        expected_session_generation=ready["session_generation"],  # type: ignore[arg-type]
        expected_state_revision=ready["state_revision"],  # type: ignore[arg-type]
    )
    scanning = _wait_phase(session, "scanning")
    assert requested["state"] in {"active-recovery-requested", "scanning"}
    assert "start:11-14" not in FakeExecutor.records
    assert FakeExecutor.records.count("adopt:scanning") == 1

    session.request_stop(
        confirmation=_confirmation(),
        command_context=_physical_context("stop"),
        expected_session_generation=scanning["session_generation"],  # type: ignore[arg-type]
        expected_state_revision=scanning["state_revision"],  # type: ignore[arg-type]
    )
    _wait_phase(session, "completed")
    assert FakeExecutor.records.count("stop") == 1


def test_reconciled_scanning_adoption_requires_latest_exact_reconciliation(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """An inherited prepared classification cannot override a newer rebind."""

    FakeExecutor.records = []
    monkeypatch.setattr(session_module, "PhysicalAcceptanceDialogueExecutor", FakeExecutor)
    coordinator = FakePhysicalCommandCoordinator()
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
    )
    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        connection_binding=_connection_binding(),
        inspection_only=True,
    )
    ready = _wait_phase(session, "connection-ready")
    verified = ready["verified_control"]
    assert isinstance(verified, dict)
    exact_connection = {
        field: verified[field]
        for field in (
            "intent_id",
            "transport_ref",
            "connection_mode",
            "target_ipv4",
            "target_port",
            "host_path_epoch",
            "control_session_id",
            "producer_generation",
        )
    }

    def reconciliation(reconciliation_id: str, kind: str) -> dict[str, object]:
        return {
            "reconciliation_id": reconciliation_id,
            "kind": kind,
            "resolution": "physical-active-observed",
            "verified_binding": {"connection": exact_connection},
            "observation": {
                "source": "explicit-read-only-reconciliation",
                "control_session_id": verified["control_session_id"],
                "host_path_epoch": verified["host_path_epoch"],
                "producer_generation": verified["producer_generation"],
                "session_state": "scanning",
                "project_bound": True,
                "init_ready": True,
                "mqtt_retained": False,
            },
        }

    coordinator.snapshot_override = {
        "status": "resolved",
        "requires_reconciliation": False,
        "resolved_active_recovery_required": True,
        "record": {
            "resolution": "start-active-observed",
            "reconciliations": [
                reconciliation("older-prepared-stop", "prepared-stop-classification"),
                reconciliation("latest-rebind", "resolved-active-rebind"),
            ],
        },
    }

    with pytest.raises(ApplicationAcceptanceError, match="does not match"):
        session.adopt_reconciled_scanning(
            reconciliation_id="older-prepared-stop",
            expected_session_generation=ready["session_generation"],  # type: ignore[arg-type]
            expected_state_revision=ready["state_revision"],  # type: ignore[arg-type]
        )

    admitted = session.adopt_reconciled_scanning(
        reconciliation_id="latest-rebind",
        expected_session_generation=ready["session_generation"],  # type: ignore[arg-type]
        expected_state_revision=ready["state_revision"],  # type: ignore[arg-type]
    )
    assert admitted["state"] in {"active-recovery-requested", "scanning"}
    scanning = _wait_phase(session, "scanning")
    session.request_stop(
        confirmation=_confirmation(),
        command_context=_physical_context("stop"),
        expected_session_generation=scanning["session_generation"],  # type: ignore[arg-type]
        expected_state_revision=scanning["state_revision"],  # type: ignore[arg-type]
    )
    _wait_phase(session, "completed")
    assert [action for _context, action, _envelope in coordinator.prepares] == ["stop"]

    with pytest.raises(ApplicationControlStateConflict):
        session.request_stop(
            confirmation=_confirmation(),
            command_context=_physical_context("stop"),
            expected_session_generation=scanning["session_generation"],  # type: ignore[arg-type]
            expected_state_revision=scanning["state_revision"],  # type: ignore[arg-type]
        )
    assert FakeExecutor.records.count("stop") == 1


def test_control_checkpoint_revision_blocks_stale_browser_continuation(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    FakeExecutor.records = []
    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        FakeExecutor,
    )
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
    )

    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
        connection_binding=_connection_binding(),
    )
    connection_ready = _wait_phase(session, "connection-ready")
    generation = connection_ready["session_generation"]
    revision = connection_ready["state_revision"]
    assert generation == 1
    assert isinstance(revision, int)

    session.enter_workspace(
        expected_session_generation=generation,
        expected_state_revision=revision,
    )
    with pytest.raises(ApplicationControlStateConflict, match="session changed"):
        session.enter_workspace(
            expected_session_generation=generation,
            expected_state_revision=revision,
        )

    workspace_ready = _wait_phase(session, "workspace-ready")
    with pytest.raises(ApplicationControlStateConflict, match="session changed"):
        session.open_project_prompt(
            expected_session_generation=generation,
            expected_state_revision=revision,
        )
    assert session.snapshot()["state"] == "workspace-ready"

    session.open_project_prompt(
        expected_session_generation=workspace_ready["session_generation"],  # type: ignore[arg-type]
        expected_state_revision=workspace_ready["state_revision"],  # type: ignore[arg-type]
    )
    _wait_phase(session, "project-ready")


def test_close_wins_over_workspace_continuation_already_in_preflight() -> None:
    validation_entered = threading.Event()
    release_validation = threading.Event()

    def validate_binding(_binding: ApplicationConnectionBinding) -> bool:
        validation_entered.set()
        assert release_validation.wait(timeout=1.0)
        return True

    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        connection_path_validator=lambda _binding: True,
        connection_binding_validator=validate_binding,
    )
    # Isolate the operator-CAS race without opening a background MQTT worker.
    # The public facade supplies these exact fields from a real snapshot.
    with session._lock:  # noqa: SLF001
        session._connection_binding = _connection_binding()  # noqa: SLF001
        session._run_generation = 1  # noqa: SLF001
        session._phase = "connection-ready"  # noqa: SLF001
        session._state_revision = 2  # noqa: SLF001

    failures: list[BaseException] = []

    def enter_workspace() -> None:
        try:
            session.enter_workspace(
                expected_session_generation=1,
                expected_state_revision=2,
            )
        except BaseException as exc:  # pragma: no branch - asserted below
            failures.append(exc)

    thread = threading.Thread(target=enter_workspace)
    thread.start()
    assert validation_entered.wait(timeout=1.0)

    closed = session.close_prestart(
        expected_session_generation=1,
        expected_state_revision=2,
    )
    release_validation.set()
    thread.join(timeout=1.0)

    assert not thread.is_alive()
    assert len(failures) == 1
    assert isinstance(failures[0], ApplicationControlStateConflict)
    assert closed["state_revision"] == 3
    assert session.snapshot()["state"] == "connection-ready"
    assert session._workspace_requested.is_set() is False  # noqa: SLF001


def test_physical_reconciliation_uses_read_only_path_not_command_authority() -> None:
    """The proof that unlocks an ambiguous edge cannot require that edge unlocked."""

    path_validations: list[ApplicationConnectionBinding] = []
    command_validations: list[ApplicationConnectionBinding] = []

    def validate_path(binding: ApplicationConnectionBinding) -> bool:
        path_validations.append(binding)
        return True

    def reject_command_authority(binding: ApplicationConnectionBinding) -> bool:
        command_validations.append(binding)
        return False

    transport = FakeTransport("192.168.1.20")
    transport.open()
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        connection_path_validator=validate_path,
        connection_binding_validator=reject_command_authority,
    )
    binding = _connection_binding()
    with session._lock:  # noqa: SLF001
        session._connection_binding = binding  # noqa: SLF001
        session._live_control_binding = FakeExecutor(transport).binding  # noqa: SLF001
        session._transport = transport  # type: ignore[assignment]  # noqa: SLF001
        session._verified_control = {"logical_device_id": "device-id"}  # noqa: SLF001
        session._phase = "connection-ready"  # noqa: SLF001

    session.validate_physical_reconciliation_binding()

    assert path_validations == [binding]
    assert command_validations == []
    with pytest.raises(ApplicationConnectionBindingLost):
        session.validate_connection_binding()
    assert command_validations == [binding]


def test_physical_reconciliation_rejects_stale_mqtt_control_proof() -> None:
    """Read-only reconciliation never outlives its retained MQTT proof."""

    transport = FakeTransport("192.168.1.20")
    transport.open()
    transport.proof_fresh = False
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        connection_path_validator=lambda _binding: True,
    )
    with session._lock:  # noqa: SLF001
        session._connection_binding = _connection_binding()  # noqa: SLF001
        session._live_control_binding = FakeExecutor(transport).binding  # noqa: SLF001
        session._transport = transport  # type: ignore[assignment]  # noqa: SLF001
        session._verified_control = {"logical_device_id": "device-id"}  # noqa: SLF001
        session._phase = "connection-ready"  # noqa: SLF001

    with pytest.raises(ApplicationControlProofExpired):
        session.validate_physical_reconciliation_binding()


def test_physical_reconciliation_rejects_host_path_drift() -> None:
    """A fresh MQTT proof cannot authorize a changed host-path epoch."""

    transport = FakeTransport("192.168.1.20")
    transport.open()
    path_validations: list[ApplicationConnectionBinding] = []

    def reject_path(binding: ApplicationConnectionBinding) -> bool:
        path_validations.append(binding)
        return False

    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        connection_path_validator=reject_path,
    )
    binding = _connection_binding()
    with session._lock:  # noqa: SLF001
        session._connection_binding = binding  # noqa: SLF001
        session._live_control_binding = FakeExecutor(transport).binding  # noqa: SLF001
        session._transport = transport  # type: ignore[assignment]  # noqa: SLF001
        session._verified_control = {"logical_device_id": "device-id"}  # noqa: SLF001
        session._phase = "connection-ready"  # noqa: SLF001

    with pytest.raises(ApplicationConnectionBindingLost):
        session.validate_physical_reconciliation_binding()
    assert path_validations == [binding]


def test_durable_prepare_failure_does_not_release_start_checkpoint(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    FakeExecutor.records = []
    monkeypatch.setattr(session_module, "PhysicalAcceptanceDialogueExecutor", FakeExecutor)
    coordinator = FakePhysicalCommandCoordinator(fail_prepare=True)
    transport = FakeTransport("192.168.1.20")
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda _host: transport,  # type: ignore[arg-type]
        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
    )
    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
        connection_binding=_connection_binding(),
    )
    _wait_phase(session, "connection-ready")
    session.enter_workspace()
    _wait_phase(session, "workspace-ready")
    session.open_project_prompt()
    _wait_phase(session, "project-ready")

    with pytest.raises(RuntimeError, match="durable prepare failed"):
        session.request_start(
            project_name="TEST001",
            confirmation=_confirmation(),
            command_context=_physical_context("start"),
            preparation_checkpoint_observer=_successful_start_checkpoint_observer,  # type: ignore[arg-type]
        )

    snapshot = session.snapshot()
    assert snapshot["state"] == "project-ready"
    assert snapshot["pending_operator_action"] == "start"
    assert session._start_requested.is_set() is False  # noqa: SLF001
    assert "start:11-14" not in FakeExecutor.records
    assert len(coordinator.prepares) == 1


def test_start_checkpoint_is_committed_before_worker_event_and_publish(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    FakeExecutor.records = []
    monkeypatch.setattr(session_module, "PhysicalAcceptanceDialogueExecutor", FakeExecutor)
    coordinator = FakePhysicalCommandCoordinator()
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
    )
    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        connection_binding=_connection_binding(),
    )
    _wait_phase(session, "connection-ready")
    session.enter_workspace()
    _wait_phase(session, "workspace-ready")
    session.open_project_prompt()
    _wait_phase(session, "project-ready")
    observed: list[str] = []

    def checkpoint(
        phase: str,
        _context: PhysicalCommandIntentContext,
        _envelope: object,
    ) -> None:
        assert phase == "prepared"
        assert len(coordinator.prepares) == 1
        assert session._start_requested.is_set() is False  # noqa: SLF001
        assert "start:11-14" not in FakeExecutor.records
        observed.append(phase)

    session.request_start(
        project_name="TEST001",
        confirmation=_confirmation(),
        command_context=_physical_context("start"),
        preparation_checkpoint_observer=checkpoint,  # type: ignore[arg-type]
    )

    assert observed == ["prepared"]
    assert session._start_requested.is_set() is True  # noqa: SLF001
    _wait_phase(session, "scanning")


def test_scanning_checkpoint_activation_linearizes_before_stop_prepare(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    FakeExecutor.records = []
    monkeypatch.setattr(session_module, "PhysicalAcceptanceDialogueExecutor", FakeExecutor)
    coordinator = FakePhysicalCommandCoordinator()
    activation_entered = threading.Event()
    release_activation = threading.Event()

    def scanning_checkpoint() -> None:
        activation_entered.set()
        assert release_activation.wait(2.0)

    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
        scanning_observer=scanning_checkpoint,
    )
    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        connection_binding=_connection_binding(),
    )
    _wait_phase(session, "connection-ready")
    session.enter_workspace()
    _wait_phase(session, "workspace-ready")
    session.open_project_prompt()
    _wait_phase(session, "project-ready")
    session.request_start(
        project_name="TEST001",
        confirmation=_confirmation(),
        command_context=_physical_context("start"),
        preparation_checkpoint_observer=_successful_start_checkpoint_observer,  # type: ignore[arg-type]
    )
    assert activation_entered.wait(2.0)
    assert session.snapshot()["state"] == "initializing"

    stop_errors: list[BaseException] = []

    def request_stop() -> None:
        try:
            session.request_stop(
                confirmation=_confirmation(),
                command_context=_physical_context("stop"),
            )
        except BaseException as exc:  # pragma: no cover - asserted below
            stop_errors.append(exc)

    stop_thread = threading.Thread(target=request_stop, daemon=True)
    stop_thread.start()
    threading.Event().wait(0.05)
    assert stop_thread.is_alive()
    assert [action for _context, action, _envelope in coordinator.prepares] == [
        "start"
    ]

    release_activation.set()
    stop_thread.join(2.0)
    assert stop_thread.is_alive() is False
    assert stop_errors == []
    assert [action for _context, action, _envelope in coordinator.prepares] == [
        "start",
        "stop",
    ]
    _wait_phase(session, "completed")


def test_failed_scanning_checkpoint_cannot_be_bypassed_by_stop_prepare(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    FakeExecutor.records = []
    monkeypatch.setattr(session_module, "PhysicalAcceptanceDialogueExecutor", FakeExecutor)
    coordinator = FakePhysicalCommandCoordinator()
    observer_calls = 0

    def failed_checkpoint() -> None:
        nonlocal observer_calls
        observer_calls += 1
        raise RuntimeError("checkpoint activation unavailable")

    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
        scanning_observer=failed_checkpoint,
    )
    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        connection_binding=_connection_binding(),
    )
    _wait_phase(session, "connection-ready")
    session.enter_workspace()
    _wait_phase(session, "workspace-ready")
    session.open_project_prompt()
    _wait_phase(session, "project-ready")
    session.request_start(
        project_name="TEST001",
        confirmation=_confirmation(),
        command_context=_physical_context("start"),
        preparation_checkpoint_observer=_successful_start_checkpoint_observer,  # type: ignore[arg-type]
    )
    _wait_phase(session, "scanning")

    with pytest.raises(ApplicationAcceptanceError, match="durably activated"):
        session.request_stop(
            confirmation=_confirmation(),
            command_context=_physical_context("stop"),
        )

    assert observer_calls == 2
    assert [action for _context, action, _envelope in coordinator.prepares] == [
        "start"
    ]
    assert session.snapshot()["state"] == "scanning"


def test_start_checkpoint_failure_settles_no_dispatch_before_cease_callback(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    FakeExecutor.records = []
    monkeypatch.setattr(session_module, "PhysicalAcceptanceDialogueExecutor", FakeExecutor)
    coordinator = FakePhysicalCommandCoordinator()
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
    )
    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        connection_binding=_connection_binding(),
    )
    _wait_phase(session, "connection-ready")
    session.enter_workspace()
    _wait_phase(session, "workspace-ready")
    session.open_project_prompt()
    _wait_phase(session, "project-ready")
    coordinator.phase_probe = lambda: str(session.snapshot()["state"])
    phases: list[str] = []

    def checkpoint(
        phase: str,
        _context: PhysicalCommandIntentContext,
        _envelope: object,
    ) -> None:
        phases.append(phase)
        if phase == "prepared":
            raise RuntimeError("checkpoint fsync failed")
        assert coordinator.resolutions == [
            ("start-not-dispatched", "project-ready")
        ]

    with pytest.raises(RuntimeError, match="checkpoint fsync failed"):
        session.request_start(
            project_name="TEST001",
            confirmation=_confirmation(),
            command_context=_physical_context("start"),
            preparation_checkpoint_observer=checkpoint,  # type: ignore[arg-type]
        )

    assert phases == ["prepared", "resolved-not-dispatched"]
    assert session.snapshot()["state"] == "project-ready"
    assert session._start_requested.is_set() is False  # noqa: SLF001
    assert "start:11-14" not in FakeExecutor.records


def test_start_checkpoint_cease_failure_is_nonretryable_settlement_error(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    FakeExecutor.records = []
    monkeypatch.setattr(session_module, "PhysicalAcceptanceDialogueExecutor", FakeExecutor)
    coordinator = FakePhysicalCommandCoordinator()
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
    )
    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        connection_binding=_connection_binding(),
    )
    _wait_phase(session, "connection-ready")
    session.enter_workspace()
    _wait_phase(session, "workspace-ready")
    session.open_project_prompt()
    _wait_phase(session, "project-ready")
    coordinator.phase_probe = lambda: str(session.snapshot()["state"])

    def checkpoint(
        phase: str,
        _context: PhysicalCommandIntentContext,
        _envelope: object,
    ) -> None:
        raise RuntimeError(
            "checkpoint prepare failed" if phase == "prepared" else "cease fsync failed"
        )

    with pytest.raises(ApplicationStartCheckpointSettlementError):
        session.request_start(
            project_name="TEST001",
            confirmation=_confirmation(),
            command_context=_physical_context("start"),
            preparation_checkpoint_observer=checkpoint,  # type: ignore[arg-type]
        )

    assert coordinator.resolutions == [
        ("start-not-dispatched", "project-ready")
    ]
    assert session.snapshot()["state"] == "project-ready"
    assert session._start_requested.is_set() is False  # noqa: SLF001
    assert "start:11-14" not in FakeExecutor.records


def test_session_generation_and_resolution_order_are_bound_to_durable_coordinator(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    FakeExecutor.records = []
    monkeypatch.setattr(session_module, "PhysicalAcceptanceDialogueExecutor", FakeExecutor)
    coordinator = FakePhysicalCommandCoordinator()
    transport = FakeTransport("192.168.1.20")
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda _host: transport,  # type: ignore[arg-type]
        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
    )
    coordinator.phase_probe = lambda: str(session.snapshot()["state"])
    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
        connection_binding=_connection_binding(),
    )
    _wait_phase(session, "connection-ready")
    assert transport.evidence_observer is coordinator
    assert len(coordinator.bindings) == 1
    runtime_binding = coordinator.bindings[0]
    assert runtime_binding.producer_generation == session._run_generation == 1  # noqa: SLF001
    assert runtime_binding.control_session_id.startswith("application-control-1-")

    session.enter_workspace()
    _wait_phase(session, "workspace-ready")
    session.open_project_prompt()
    _wait_phase(session, "project-ready")
    session.request_start(
        project_name="TEST001",
        confirmation=_confirmation(),
        command_context=_physical_context("start"),
        preparation_checkpoint_observer=_successful_start_checkpoint_observer,  # type: ignore[arg-type]
    )
    _wait_phase(session, "scanning")
    session.request_stop(
        confirmation=_confirmation(),
        command_context=_physical_context("stop"),
    )
    _wait_phase(session, "completed")

    assert [action for _context, action, _envelope in coordinator.prepares] == [
        "start",
        "stop",
    ]
    assert coordinator.resolutions == [
        ("start", "initializing"),
        ("stop", "awaiting-standby-confirmation"),
    ]


def test_retired_identity_under_new_transport_is_rejected_after_device_info_only(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A new BLE UUID cannot move a retired K1 past read-only DeviceInfo."""

    old_transport_ref = "A161D9D5-C352-1069-D430-5FB0BC13F7F9"
    new_transport_ref = "B272E0E6-D463-2170-E541-6FC1CD24A8A0"
    identity = PhysicalCommandIdentity(
        vendor_device_id_sha256=hashlib.sha256(
            RETIRED_VENDOR_DEVICE_ID.encode()
        ).hexdigest(),
        device_serial_sha256=hashlib.sha256(
            RETIRED_DEVICE_SERIAL.encode()
        ).hexdigest(),
    )
    old_connection = PhysicalCommandConnectionBinding(
        intent_id="retired-intent-1",
        transport_ref=old_transport_ref,
        connection_mode="bridge",
        target_ipv4="192.168.68.51",
        target_port=1883,
        host_path_epoch=7,
        control_session_id="retired-control-session-1",
        producer_generation=11,
    )
    baseline = PhysicalCommandStatusEvidence(
        source="live-control-session",
        vendor_device_id_sha256=identity.vendor_device_id_sha256,
        device_serial_sha256=identity.device_serial_sha256,
        control_session_id=old_connection.control_session_id,
        host_path_epoch=old_connection.host_path_epoch,
        producer_generation=old_connection.producer_generation,
        session_state="ready",
        session_state_code=300,
        project_bound=False,
        project_id_sha256=None,
        init_ready=False,
        status_message_sha256="d" * 64,
        mqtt_retained=False,
        observed_at_utc="2026-08-10T18:00:00.000Z",
    )
    monkeypatch.setenv("MISSIONCORE_DATA_DIR", str(tmp_path / "private-data"))
    ledger = PhysicalCommandLedger(tmp_path / "repository")
    operation_id = "retired-start-before-bootstrap-ordering-test"
    ledger.prepare(
        operation_id=operation_id,
        parent_operation_id=None,
        acquisition_id="retired-acquisition-before-bootstrap-ordering-test",
        action="start",
        identity=identity,
        connection=old_connection,
        compatibility_profile_id=COMPATIBILITY_PROFILE_ID,
        payload_sha256="e" * 64,
        baseline_status=baseline,
    )
    ledger.mark_dispatching(operation_id)
    unresolved = ledger.snapshot().record
    assert unresolved is not None
    ledger.retire_unavailable_target(
        retirement_id="retired-bootstrap-ordering-test",
        expected_operation_id=operation_id,
        expected_revision=unresolved.revision,
        expected_transport_ref=old_transport_ref,
        reason="device-permanently-unavailable-or-replaced",
    )
    durable_bytes = ledger.path.read_bytes()

    @dataclass
    class OrderingTransportSnapshot:
        state: str
        publish_attempts: int
        correlated_responses: int

        def as_dict(self) -> dict[str, object]:
            return {
                "state": self.state,
                "publish_attempts": self.publish_attempts,
                "qos2_completions": self.correlated_responses,
                "correlated_responses": self.correlated_responses,
                "ignored_known_responses": 0,
                "late_known_responses": 0,
                "automatic_retry": False,
                "automatic_reconnect": False,
            }

    class OrderingTransport(FakeTransport):
        def __init__(self, host: str) -> None:
            super().__init__(host)
            self.batches: list[tuple[str, ...]] = []
            self.publish_attempts = 0
            self.correlated_responses = 0

        def exchange_batch_once(
            self,
            envelopes: Sequence[OneShotPublishEnvelope],
            *,
            required_response_operation_keys: Collection[str],
        ) -> dict[str, bytes]:
            operations = tuple(envelope.operation_key for envelope in envelopes)
            self.batches.append(operations)
            self.publish_attempts += len(envelopes)
            responses: dict[str, bytes] = {}
            for operation_key in required_response_operation_keys:
                _, ordinal_text, message_type = operation_key.split(":", 2)
                ordinal = int(ordinal_text)
                session_id = (
                    f":{message_type}"
                    if ordinal <= 3
                    else f"{RETIRED_VENDOR_DEVICE_ID}:{message_type}"
                )
                if message_type == "DeviceConfigRequest":
                    session_id += ":Publish_Proto_DeviceConfig_SetTime"
                responses[operation_key] = (
                    _test_device_info_response(session_id)
                    if message_type == "DeviceInfoRequest"
                    else _test_generic_response(session_id)
                )
                self.correlated_responses += 1
            return responses

        def snapshot(self) -> OrderingTransportSnapshot:
            return OrderingTransportSnapshot(
                self.state,
                self.publish_attempts,
                self.correlated_responses,
            )

        def maintain_open_for(
            self,
            _duration_seconds: float,
            *,
            allowed_response_topics: Collection[str] = (),
        ) -> None:
            del allowed_response_topics

        def scan_initialization_complete(self, _binding: object) -> bool:
            return False

        def pre_start_ready(self, _binding: object) -> bool:
            return True

        def standby_complete(self, _binding: object) -> bool:
            return False

        def validate_bound_status(self, _binding: object) -> None:
            return None

    transport = OrderingTransport("192.168.68.51")
    coordinator = LedgerPhysicalCommandCoordinator(ledger)
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda _host: transport,  # type: ignore[arg-type]
        physical_command_coordinator=coordinator,
    )
    session.open(
        host="192.168.68.51",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
        connection_binding=ApplicationConnectionBinding(
            intent_id="fresh-intent-for-retired-identity",
            transport_ref=new_transport_ref,
            host_path_epoch=8,
            target_ipv4="192.168.68.51",
            target_port=1883,
            connection_mode="bridge",
        ),
    )
    failed = _wait_phase(session, "failed")
    worker = session._thread  # noqa: SLF001
    assert worker is not None
    worker.join(timeout=2.0)
    assert not worker.is_alive()

    assert transport.batches == [("bootstrap:1:DeviceInfoRequest",)]
    emitted = tuple(operation for batch in transport.batches for operation in batch)
    assert not any("DeviceConfigRequest" in operation for operation in emitted)
    assert not any(operation.startswith("dialogue:") for operation in emitted)
    assert not any(operation.startswith("modeling:") for operation in emitted)
    assert failed["verified_control"] is None
    assert failed["can_enter_workspace"] is False
    assert failed["can_prepare_project"] is False
    assert failed["can_start"] is False
    assert failed["failure"]["reason_code"] == (  # type: ignore[index]
        "physical-command-reconciliation-required"
    )
    assert ledger.path.read_bytes() == durable_bytes


def test_start_preflight_fails_closed_when_control_proof_is_stale(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    FakeExecutor.records = []
    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        FakeExecutor,
    )
    transport = FakeTransport("192.168.1.20")
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda _host: transport,  # type: ignore[arg-type]
        epoch_seconds=lambda: 1_752_680_000,
    )
    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
        connection_binding=_connection_binding(),
    )
    _wait_phase(session, "connection-ready")
    session.enter_workspace()
    _wait_phase(session, "workspace-ready")
    session.open_project_prompt()
    _wait_phase(session, "project-ready")

    transport.proof_fresh = False

    session.request_start(project_name="TEST001", confirmation=_confirmation())
    failed = _wait_phase(session, "failed")
    failure = failed["failure"]
    assert isinstance(failure, dict)
    assert failure["reason_code"] == ApplicationControlProofExpired.reason_code
    assert failure["modeling_command_attempted"] is False
    assert "start:11-14" not in FakeExecutor.records


@pytest.mark.parametrize(
    ("guarded_stage", "forbidden_record", "outcome_unknown"),
    [
        ("workspace", "workspace:7", False),
        ("project", "project:8-10", False),
        ("start", "start:11-14", False),
        ("stop", "stop", True),
    ],
)
def test_stale_connection_binding_blocks_each_pending_canonical_command(
    monkeypatch: pytest.MonkeyPatch,
    guarded_stage: str,
    forbidden_record: str,
    outcome_unknown: bool,
) -> None:
    FakeExecutor.records = []
    binding_current = True
    allow_one_stale_preflight = False

    def validate_binding(_binding: ApplicationConnectionBinding) -> bool:
        nonlocal allow_one_stale_preflight
        if allow_one_stale_preflight:
            allow_one_stale_preflight = False
            return True
        return binding_current

    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        FakeExecutor,
    )
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
        connection_binding_validator=validate_binding,
    )
    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
        connection_binding=_connection_binding(),
    )
    _wait_phase(session, "connection-ready")

    if guarded_stage != "workspace":
        session.enter_workspace()
        _wait_phase(session, "workspace-ready")
    if guarded_stage not in {"workspace", "project"}:
        session.open_project_prompt()
        _wait_phase(session, "project-ready")
    if guarded_stage == "stop":
        session.request_start(project_name="TEST001", confirmation=_confirmation())
        _wait_phase(session, "scanning")

    binding_current = False
    if guarded_stage == "workspace":
        allow_one_stale_preflight = True
        session.enter_workspace()
    elif guarded_stage == "project":
        allow_one_stale_preflight = True
        session.open_project_prompt()
    elif guarded_stage == "start":
        session.request_start(project_name="TEST001", confirmation=_confirmation())
    else:
        session.request_stop(confirmation=_confirmation())

    failed = _wait_phase(session, "failed")
    worker = session._thread  # noqa: SLF001
    assert worker is not None
    worker.join(timeout=2.0)
    assert not worker.is_alive()
    failed = session.snapshot()

    assert forbidden_record not in FakeExecutor.records
    assert failed["outcome_unknown"] is outcome_unknown
    failure = failed["failure"]
    assert isinstance(failure, dict)
    assert failure["reason_code"] == "application-connection-binding-lost"
    assert failure["modeling_command_attempted"] is False
    assert failure["safe_to_retry"] is (not outcome_unknown)


def test_binding_is_revalidated_after_correlated_response_before_phase_promotion(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    FakeExecutor.records = []
    binding_current = True

    class RouteChangingExecutor(FakeExecutor):
        def run_workspace_entry_stage(
            self,
            orchestrator: object,
            checkpoint: object,
            *,
            dispatch_guard: Callable[[], None] | None = None,
        ) -> LiveDeviceControlBinding:
            nonlocal binding_current
            result = super().run_workspace_entry_stage(
                orchestrator,
                checkpoint,
                dispatch_guard=dispatch_guard,
            )
            binding_current = False
            return result

    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        RouteChangingExecutor,
    )
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
        connection_binding_validator=lambda _binding: binding_current,
    )
    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
        connection_binding=_connection_binding(),
    )
    _wait_phase(session, "connection-ready")

    session.enter_workspace()
    failed = _wait_phase(session, "failed")

    assert "workspace:7" in FakeExecutor.records
    assert failed["state"] != "workspace-ready"
    assert failed["failure"]["reason_code"] == (  # type: ignore[index]
        "application-connection-binding-lost"
    )


def test_prestart_failure_is_reported_and_requires_a_new_operator_click(
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    caplog.set_level(logging.ERROR)

    class BootstrapFailureExecutor(FakeExecutor):
        def run_read_only_inspection_stage(
            self,
            _orchestrator: object,
        ) -> LiveDeviceControlBinding:
            raise RuntimeError("bootstrap correlation failed")

        def snapshot(self) -> dict[str, object]:
            return {
                "dialogue_stage": "connection",
                "start_attempted": False,
                "stop_attempted": False,
                "response_evidence": [],
                "automatic_retry": False,
            }

    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        BootstrapFailureExecutor,
    )
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
    )

    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
    )
    _wait_phase(session, "failed")
    failed_thread = session._thread  # noqa: SLF001
    assert failed_thread is not None
    failed_thread.join(timeout=2.0)
    assert not failed_thread.is_alive()
    failed = session.snapshot()

    assert failed["outcome_unknown"] is False
    assert failed["can_open"] is True
    assert failed["automatic_retry"] is False
    assert failed["failure"] == {
        "code": "RuntimeError",
        "reason_code": "unexpected_runtime_error",
        "message": "bootstrap correlation failed",
        "failed_phase": "connecting",
        "dialogue_stage": "connection",
        "transport_state": "ready",
        "publish_attempts": 0,
        "stop_publish_attempts": None,
        "qos2_completions": 0,
        "correlated_responses": 0,
        "ignored_known_responses": 0,
            "late_known_responses": 0,
            "modeling_command_attempted": False,
            "stop_command_attempted": False,
            "diagnostic_snapshot_unavailable": [],
        "diagnostic_evidence_unavailable": [],
        "correlation_failure": None,
        "compatibility_failure": None,
        "safe_to_retry": True,
    }
    assert "reason_code=unexpected_runtime_error" in caplog.text


def test_correlated_read_only_profile_mismatch_allows_only_a_fresh_explicit_attempt(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    class ProfileMismatchError(RuntimeError):
        reason_code = "compatibility_profile_mismatch"

    @dataclass
    class CorrelatedReadSnapshot:
        state: str

        def as_dict(self) -> dict[str, object]:
            return {
                "state": self.state,
                "publish_attempts": 1,
                "qos2_completions": 1,
                "correlated_responses": 1,
                "ignored_known_responses": 0,
                "late_known_responses": 0,
                "automatic_retry": False,
                "automatic_reconnect": False,
            }

    class CorrelatedReadTransport(FakeTransport):
        def snapshot(self) -> CorrelatedReadSnapshot:
            return CorrelatedReadSnapshot(self.state)

    class ProfileMismatchExecutor(FakeExecutor):
        def run_read_only_inspection_stage(
            self,
            _orchestrator: object,
        ) -> LiveDeviceControlBinding:
            raise ProfileMismatchError("live DeviceInfo profile mismatch")

        def snapshot(self) -> dict[str, object]:
            return {
                "dialogue_stage": "new",
                "start_attempted": False,
                "stop_attempted": False,
                "response_evidence": [],
                "correlation_failure": None,
                "compatibility_failure": {
                    "operation_key": "bootstrap:1:DeviceInfoRequest",
                    "reason_code": "compatibility_profile_mismatch",
                },
                "automatic_retry": False,
            }

    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        ProfileMismatchExecutor,
    )
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: CorrelatedReadTransport(host),  # type: ignore[arg-type]
    )

    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
    )
    failed_thread = session._thread  # noqa: SLF001
    assert failed_thread is not None
    failed_thread.join(timeout=2.0)
    assert not failed_thread.is_alive()

    failed = session.snapshot()
    assert failed["state"] == "failed"
    assert failed["outcome_unknown"] is False
    assert failed["automatic_retry"] is False
    assert failed["can_open"] is True
    failure = failed["failure"]
    assert isinstance(failure, dict)
    assert failure["reason_code"] == "compatibility_profile_mismatch"
    assert failure["modeling_command_attempted"] is False
    assert failure["safe_to_retry"] is True


def test_new_explicit_session_waits_for_old_worker_transport_retirement(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    old_close_entered = threading.Event()
    release_old_close = threading.Event()
    transports: list[TaggedTransport] = []

    @dataclass
    class TaggedTransportSnapshot:
        state: str
        ready: bool
        tag: str

        def as_dict(self) -> dict[str, object]:
            return {
                "state": self.state,
                "tag": self.tag,
                "publish_attempts": 0,
                "qos2_completions": 0,
                "correlated_responses": 0,
                "ignored_known_responses": 0,
                "late_known_responses": 0,
                "latest_device_session_state": "ready" if self.ready else "scanning",
                "latest_device_project_bound": not self.ready,
                "automatic_retry": False,
                "automatic_reconnect": False,
            }

    class TaggedTransport(FakeTransport):
        def __init__(self, host: str, tag: str) -> None:
            super().__init__(host)
            self.tag = tag
            self.close_calls = 0

        def close(self) -> None:
            if self.tag == "generation-1":
                old_close_entered.set()
                assert release_old_close.wait(timeout=2.0)
            self.close_calls += 1
            super().close()

        def snapshot(self) -> TaggedTransportSnapshot:
            return TaggedTransportSnapshot(self.state, self.ready, self.tag)

    class RacingExecutor(FakeExecutor):
        instances = 0

        def __init__(self, transport: TaggedTransport) -> None:
            super().__init__(transport)
            type(self).instances += 1
            self.instance = type(self).instances

        def run_read_only_inspection_stage(
            self,
            orchestrator: object,
        ) -> LiveDeviceControlBinding:
            if self.instance == 1:
                raise RuntimeError("first generation failed before publish")
            return super().run_read_only_inspection_stage(orchestrator)

    def transport_factory(host: str) -> TaggedTransport:
        transport = TaggedTransport(host, f"generation-{len(transports) + 1}")
        transports.append(transport)
        return transport

    FakeExecutor.records = []
    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        RacingExecutor,
    )
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=transport_factory,  # type: ignore[arg-type]
    )

    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
    )
    _wait_phase(session, "failed")
    assert old_close_entered.wait(timeout=2.0)
    old_thread = session._thread  # noqa: SLF001
    assert old_thread is not None

    # The failure is safe to retry at the protocol level, but the old worker
    # still owns its socket. A new explicit click must not construct or open a
    # second transport until close() completes and the worker exits.
    retiring = session.snapshot()
    assert retiring["failure"]["safe_to_retry"] is True  # type: ignore[index]
    assert retiring["can_open"] is False
    with pytest.raises(
        session_module.ApplicationAcceptanceError,
        match="already open or requires manual recovery",
    ):
        session.open(
            host="192.168.1.20",
            timezone_name="Europe/Moscow",
            confirmation=_confirmation(),
        )
    assert len(transports) == 1
    assert transports[0].close_calls == 0

    release_old_close.set()
    old_thread.join(timeout=2.0)
    assert not old_thread.is_alive()
    retired = session.snapshot()
    assert retired["state"] == "failed"
    assert retired["can_open"] is True
    assert transports[0].close_calls == 1

    # Only a fresh explicit action after retirement creates generation 2.
    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
    )
    new_thread = session._thread  # noqa: SLF001
    assert new_thread is not None
    current = _wait_phase(session, "connection-ready")
    assert current["transport"]["tag"] == "generation-2"  # type: ignore[index]
    assert session._transport is transports[1]  # noqa: SLF001
    assert transports[1].close_calls == 0

    # Complete the second synthetic session so no background waiter remains.
    session.enter_workspace()
    _wait_phase(session, "workspace-ready")
    session.open_project_prompt()
    _wait_phase(session, "project-ready")
    session.request_start(project_name="TEST002", confirmation=_confirmation())
    _wait_phase(session, "scanning")
    session.request_stop(confirmation=_confirmation())
    _wait_phase(session, "completed")
    new_thread.join(timeout=2.0)
    assert not new_thread.is_alive()
    assert transports[1].close_calls == 1


def test_start_outcome_unknown_blocks_reopen_even_when_transport_is_closed(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    class StartFailureExecutor(FakeExecutor):
        start_attempted = False

        def execute_canonical_start(self, *_args: object, **_kwargs: object) -> object:
            type(self).start_attempted = True
            raise session_module.ApplicationCommandOutcomeUnknown(
                "START response was not correlated"
            )

        def snapshot(self) -> dict[str, object]:
            return {
                "dialogue_stage": "start-attempted",
                "start_attempted": type(self).start_attempted,
                "stop_attempted": False,
                "response_evidence": [],
                "automatic_retry": False,
            }

    FakeExecutor.records = []
    StartFailureExecutor.start_attempted = False
    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        StartFailureExecutor,
    )
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
    )

    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
    )
    _wait_phase(session, "connection-ready")
    session.enter_workspace()
    _wait_phase(session, "workspace-ready")
    session.open_project_prompt()
    _wait_phase(session, "project-ready")
    session.request_start(project_name="TEST001", confirmation=_confirmation())
    failed = _wait_phase(session, "failed")

    assert failed["outcome_unknown"] is True
    assert failed["can_open"] is False
    assert failed["failure"]["modeling_command_attempted"] is True  # type: ignore[index]
    assert failed["failure"]["safe_to_retry"] is False  # type: ignore[index]
    with pytest.raises(
        session_module.ApplicationAcceptanceError,
        match="cannot be retired",
    ):
        session.retire_for_network_change()


def test_post_start_read_only_timeout_preserves_durable_active_proof_without_stop_authority(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    class PostStartRefreshFailureExecutor(FakeExecutor):
        start_attempted = False
        start_active_confirmed = False

        def execute_canonical_start(self, *_args: object, **kwargs: object) -> object:
            type(self).start_attempted = True
            self.records.append("start:11-12")
            self.transport.ready = False
            observer = kwargs.get("start_active_observer")
            assert callable(observer)
            observer()
            type(self).start_active_confirmed = True
            self.records.append("refresh:13-14-timeout")
            raise ApplicationMqttTransportError(
                "post-START read-only response timed out after host wake",
                reason_code="mqtt_response_timeout",
            )

        def snapshot(self) -> dict[str, object]:
            return {
                "dialogue_stage": (
                    "start-active-observed"
                    if type(self).start_active_confirmed
                    else "start-attempted"
                ),
                "start_attempted": type(self).start_attempted,
                "start_active_confirmed": type(self).start_active_confirmed,
                "start_complete": False,
                "stop_attempted": False,
                "response_evidence": [],
                "automatic_retry": False,
            }

    FakeExecutor.records = []
    PostStartRefreshFailureExecutor.start_attempted = False
    PostStartRefreshFailureExecutor.start_active_confirmed = False
    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        PostStartRefreshFailureExecutor,
    )
    coordinator = FakePhysicalCommandCoordinator()
    scanning_observed = threading.Event()
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
        scanning_observer=scanning_observed.set,
    )
    coordinator.phase_probe = lambda: str(session.snapshot()["state"])

    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        connection_binding=_connection_binding(),
    )
    _wait_phase(session, "connection-ready")
    session.enter_workspace()
    _wait_phase(session, "workspace-ready")
    session.open_project_prompt()
    _wait_phase(session, "project-ready")
    session.request_start(
        project_name="SLEEP_WAKE",
        confirmation=_confirmation(),
        command_context=_physical_context("start"),
        preparation_checkpoint_observer=_successful_start_checkpoint_observer,  # type: ignore[arg-type]
    )
    failed = _wait_phase(session, "failed")

    assert coordinator.resolutions == [("start", "initializing")]
    assert scanning_observed.is_set()
    assert failed["can_stop"] is False
    assert failed["outcome_unknown"] is True
    assert failed["failure"]["reason_code"] == "mqtt_response_timeout"  # type: ignore[index]
    assert failed["dialogue"]["start_active_confirmed"] is True  # type: ignore[index]
    assert FakeExecutor.records[-2:] == [
        "start:11-12",
        "refresh:13-14-timeout",
    ]


@pytest.mark.parametrize(
    ("disconnect_error", "expected_reason_code"),
    [
        (
            session_module.ApplicationCommandOutcomeUnknown(
                "control MQTT network loop returned an error",
                reason_code="mqtt_network_loop_failed",
            ),
            "mqtt_network_loop_failed",
        ),
        (
            session_module.ApplicationConnectionBindingLost(
                "Mac Wi-Fi route changed before READY confirmation"
            ),
            "application-connection-binding-lost",
        ),
    ],
    ids=["mqtt-network-loop-lost", "host-route-binding-lost"],
)
def test_acknowledged_stop_disconnect_allows_only_explicit_network_change(
    monkeypatch: pytest.MonkeyPatch,
    disconnect_error: Exception,
    expected_reason_code: str,
) -> None:
    @dataclass
    class StopDisconnectTransportSnapshot:
        state: str

        def as_dict(self) -> dict[str, object]:
            return {
                "state": self.state,
                "publish_attempts": 15,
                "qos2_completions": 15,
                "correlated_responses": 13,
                "ignored_known_responses": 304,
                "late_known_responses": 0,
                "device_status_reports": 52,
                "latest_device_session_state": "scan_stopping",
                "latest_device_project_bound": True,
                "latest_system_error_code": None,
                "last_loop_result_code": 7,
                "last_loop_result_name": "The connection was lost.",
                "last_loop_phase": "maintain-open",
                "automatic_retry": False,
                "automatic_reconnect": False,
            }

    class StopDisconnectTransport(FakeTransport):
        def snapshot(self) -> StopDisconnectTransportSnapshot:
            return StopDisconnectTransportSnapshot(self.state)

    class StopDisconnectExecutor(FakeExecutor):
        def __init__(self, transport: StopDisconnectTransport) -> None:
            super().__init__(transport)
            self.stop_attempted = False
            self.stop_complete = False

        def execute_canonical_start(self, *_args: object, **_kwargs: object) -> object:
            self.records.append("start:11-14")
            return object()

        def execute_canonical_stop(self, *_args: object, **_kwargs: object) -> object:
            self.records.append("stop")
            self.stop_attempted = True
            self.stop_complete = True
            return object()

        def maintain_post_stop_until_standby(self) -> None:
            self.records.append("wait:device-standby")
            self.transport.state = "poisoned"
            raise disconnect_error

        def snapshot(self) -> dict[str, object]:
            return {
                "dialogue_stage": ("stop-acknowledged" if self.stop_complete else "test-stage"),
                "start_attempted": True,
                "start_complete": True,
                "stop_attempted": self.stop_attempted,
                "stop_complete": self.stop_complete,
                "response_evidence": [],
                "automatic_retry": False,
            }

    FakeExecutor.records = []
    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        StopDisconnectExecutor,
    )
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: StopDisconnectTransport(host),  # type: ignore[arg-type]
    )

    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
    )
    _wait_phase(session, "connection-ready")
    session.enter_workspace()
    _wait_phase(session, "workspace-ready")
    session.open_project_prompt()
    _wait_phase(session, "project-ready")
    session.request_start(project_name="TEST001", confirmation=_confirmation())
    _wait_phase(session, "scanning")
    session.request_stop(confirmation=_confirmation())
    failed = _wait_phase(session, "failed")
    failed_thread = session._thread  # noqa: SLF001
    assert failed_thread is not None
    failed_thread.join(timeout=2.0)
    assert not failed_thread.is_alive()
    failed = session.snapshot()

    assert failed["can_open"] is False
    assert failed["outcome_unknown"] is True
    failure = failed["failure"]
    assert isinstance(failure, dict)
    assert failure["reason_code"] == expected_reason_code
    assert failure["safe_to_retry"] is False
    assert failure["network_change_admissible"] is True
    assert failure["network_change_reconciliation"] == {
        "device_session_state": "scan_stopping",
        "device_project_bound": True,
        "system_error_code": None,
        "stop_complete": True,
        "standby_confirmed": False,
        "decision": "explicit-network-change-only-after-acknowledged-stop",
        "automatic_retry": False,
    }

    retired = session.retire_for_network_change()
    assert retired["state"] == "idle"
    assert retired["failure"] is None
    assert retired["can_open"] is True


def test_stop_dispatch_deadline_is_definite_zero_publish_not_unknown(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    deadline_checks = 0
    expired = False

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

    class DeadlineExecutor(FakeExecutor):
        def execute_canonical_start(self, *_args: object, **_kwargs: object) -> object:
            result = super().execute_canonical_start(*_args, **_kwargs)
            self.transport.publish_attempts = 6
            return result

        def execute_canonical_stop(self, *_args: object, **kwargs: object) -> object:
            nonlocal expired
            predicate = kwargs.get("dispatch_admission_deadline_reached")
            assert callable(predicate)
            expired = True
            assert predicate() is True
            self.transport.state = "failed"
            raise ApplicationMqttTransportError(
                "control command dispatch deadline expired before publish admission",
                reason_code="physical-command-dispatch-deadline-expired",
            )

        def snapshot(self) -> dict[str, object]:
            return {
                "dialogue_stage": "stop-requested",
                "start_attempted": True,
                "stop_attempted": False,
                "stop_complete": False,
                "automatic_retry": False,
            }

    FakeExecutor.records = []
    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        DeadlineExecutor,
    )
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
    )
    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
    )
    _wait_phase(session, "connection-ready")
    session.enter_workspace()
    _wait_phase(session, "workspace-ready")
    session.open_project_prompt()
    _wait_phase(session, "project-ready")
    session.request_start(project_name="TEST001", confirmation=_confirmation())
    _wait_phase(session, "scanning")

    session.request_stop(
        confirmation=_confirmation(),
        dispatch_admission_deadline_reached=deadline_reached,
    )
    failed = _wait_phase(session, "failed")

    assert deadline_checks >= 4
    assert failed["outcome_unknown"] is False
    failure = failed["failure"]
    assert isinstance(failure, dict)
    assert failure["reason_code"] == "physical-command-dispatch-deadline-expired"
    assert failure["publish_attempts"] == 6
    assert failure["stop_publish_attempts"] == 0
    assert failure["modeling_command_attempted"] is True
    assert failure["stop_command_attempted"] is False
    assert failure["diagnostic_evidence_unavailable"] == []


def test_stop_deadline_expiring_during_durable_prepare_settles_exact_no_dispatch(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    expired = False

    class ExpiringStopPrepareCoordinator(FakePhysicalCommandCoordinator):
        def prepare(
            self,
            context: PhysicalCommandIntentContext,
            *,
            action: str,
            envelope: object,
        ) -> None:
            nonlocal expired
            super().prepare(context, action=action, envelope=envelope)
            if action == "stop":
                expired = True

    FakeExecutor.records = []
    monkeypatch.setattr(session_module, "PhysicalAcceptanceDialogueExecutor", FakeExecutor)
    coordinator = ExpiringStopPrepareCoordinator()
    transport = FakeTransport("192.168.1.20")
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda _host: transport,  # type: ignore[arg-type]
        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
    )
    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        connection_binding=_connection_binding(),
    )
    _wait_phase(session, "connection-ready")
    session.enter_workspace()
    _wait_phase(session, "workspace-ready")
    session.open_project_prompt()
    _wait_phase(session, "project-ready")
    session.request_start(
        project_name="TEST001",
        confirmation=_confirmation(),
        command_context=_physical_context("start"),
        preparation_checkpoint_observer=_successful_start_checkpoint_observer,
    )
    scanning = _wait_phase(session, "scanning")
    publish_attempts_before = transport.publish_attempts

    with pytest.raises(ApplicationMqttTransportError) as raised:
        session.request_stop(
            confirmation=_confirmation(),
            command_context=_physical_context("stop"),
            dispatch_admission_deadline_reached=lambda: expired,
            expected_session_generation=scanning["session_generation"],  # type: ignore[arg-type]
            expected_state_revision=scanning["state_revision"],  # type: ignore[arg-type]
        )

    assert raised.value.reason_code == "physical-command-dispatch-deadline-expired"
    assert session.snapshot()["state"] == "scanning"
    assert session._stop_requested.is_set() is False  # noqa: SLF001
    assert transport.publish_attempts - publish_attempts_before == 0
    assert [action for _context, action, _envelope in coordinator.prepares] == [
        "start",
        "stop",
    ]
    assert coordinator.resolutions[-1][0] == "stop-not-dispatched"
    assert FakeExecutor.records.count("stop") == 0


def test_preadmission_deadline_rejects_stop_before_validation_or_prepare() -> None:
    coordinator = FakePhysicalCommandCoordinator()
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
    )

    with pytest.raises(ApplicationMqttTransportError) as raised:
        session.request_stop(
            confirmation=_confirmation(),
            command_context=_physical_context("stop"),
            dispatch_admission_deadline_reached=lambda: True,
        )

    assert raised.value.reason_code == "physical-command-dispatch-deadline-expired"
    assert coordinator.prepares == []
    assert session.snapshot()["state"] == "idle"


def test_prestart_loop_failure_allows_only_fresh_explicit_retry_after_ready_status(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    @dataclass
    class ReconciledTransportSnapshot:
        state: str

        def as_dict(self) -> dict[str, object]:
            return {
                "state": self.state,
                "publish_attempts": 4,
                "qos2_completions": 4,
                "correlated_responses": 3,
                "ignored_known_responses": 0,
                "late_known_responses": 0,
                "device_status_reports": 1,
                "latest_device_session_state": "ready",
                "latest_device_project_bound": False,
                "latest_device_init_ready": False,
                "latest_system_error_code": None,
                "last_loop_result_code": 7,
                "last_loop_result_name": "The connection was lost.",
                "last_loop_phase": "post-publish-drain",
                "automatic_retry": False,
                "automatic_reconnect": False,
            }

    class ReconciledTransport(FakeTransport):
        def snapshot(self) -> ReconciledTransportSnapshot:
            return ReconciledTransportSnapshot(self.state)

    class PrestartLoopFailureExecutor(FakeExecutor):
        def complete_connection_stage(
            self,
            _orchestrator: object,
            *,
            expected_binding: LiveDeviceControlBinding,
        ) -> LiveDeviceControlBinding:
            assert expected_binding == self.binding
            raise session_module.ApplicationCommandOutcomeUnknown(
                "control MQTT network loop returned an error",
                reason_code="mqtt_network_loop_failed",
            )

    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        PrestartLoopFailureExecutor,
    )
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: ReconciledTransport(host),  # type: ignore[arg-type]
    )

    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
    )
    failed_thread = session._thread  # noqa: SLF001
    assert failed_thread is not None
    failed_thread.join(timeout=2.0)
    assert not failed_thread.is_alive()
    failed = session.snapshot()

    assert failed["state"] == "failed"
    assert failed["outcome_unknown"] is True
    assert failed["automatic_retry"] is False
    assert failed["can_open"] is True
    failure = failed["failure"]
    assert isinstance(failure, dict)
    assert failure["reason_code"] == "mqtt_network_loop_failed"
    assert failure["modeling_command_attempted"] is False
    assert failure["safe_to_retry"] is True
    assert failure["status_reconciliation"] == {
        "device_session_state": "ready",
        "device_project_bound": False,
        "system_error_code": None,
        "decision": "safe-explicit-prestart-retry",
        "automatic_retry": False,
    }


def test_unavailable_transport_snapshot_after_publish_blocks_reopen(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    executor_entered = threading.Event()
    release_failure = threading.Event()

    class SnapshotFailureTransport(FakeTransport):
        def __init__(self, host: str) -> None:
            super().__init__(host)
            self.simulated_publish_attempts = 0
            self.snapshot_unavailable = False

        def snapshot(self) -> FakeTransportSnapshot:
            if self.snapshot_unavailable:
                raise RuntimeError("simulated transport snapshot failure")
            return super().snapshot()

    class PublishedThenFailedExecutor(FakeExecutor):
        def complete_connection_stage(
            self,
            _orchestrator: object,
            *,
            expected_binding: LiveDeviceControlBinding,
        ) -> LiveDeviceControlBinding:
            assert expected_binding == self.binding
            executor_entered.set()
            assert release_failure.wait(timeout=2.0)
            assert isinstance(self.transport, SnapshotFailureTransport)
            self.transport.simulated_publish_attempts = 1
            self.transport.snapshot_unavailable = True
            raise RuntimeError("bootstrap failed after simulated publish")

    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        PublishedThenFailedExecutor,
    )
    transport = SnapshotFailureTransport("192.168.1.20")
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda _host: transport,  # type: ignore[arg-type]
    )

    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
    )
    assert executor_entered.wait(timeout=2.0)
    failed_thread = session._thread  # noqa: SLF001
    assert failed_thread is not None
    release_failure.set()
    failed_thread.join(timeout=2.0)
    assert not failed_thread.is_alive()
    failed = session.snapshot()

    assert transport.simulated_publish_attempts == 1
    assert failed["state"] == "failed"
    assert failed["outcome_unknown"] is True
    assert failed["can_open"] is False
    failure = failed["failure"]
    assert isinstance(failure, dict)
    assert failure["transport_state"] is None
    assert failure["publish_attempts"] is None
    assert failure["diagnostic_snapshot_unavailable"] == ["transport"]
    assert failure["diagnostic_evidence_unavailable"] == ["transport.publish_attempts"]
    assert failure["modeling_command_attempted"] is False
    assert failure["safe_to_retry"] is False
    with pytest.raises(
        session_module.ApplicationAcceptanceError,
        match="already open or requires manual recovery",
    ):
        session.open(
            host="192.168.1.20",
            timezone_name="Europe/Moscow",
            confirmation=_confirmation(),
        )


def test_unavailable_dialogue_snapshot_is_outcome_unknown(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    class DialogueSnapshotFailureExecutor(FakeExecutor):
        def run_read_only_inspection_stage(
            self,
            _orchestrator: object,
        ) -> LiveDeviceControlBinding:
            raise RuntimeError("bootstrap failed with unavailable dialogue evidence")

        def snapshot(self) -> dict[str, object]:
            raise RuntimeError("simulated dialogue snapshot failure")

    monkeypatch.setattr(
        session_module,
        "PhysicalAcceptanceDialogueExecutor",
        DialogueSnapshotFailureExecutor,
    )
    session = InteractiveApplicationControlSession(
        FakeAuthorityLoader(),
        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
    )

    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
    )
    failed_thread = session._thread  # noqa: SLF001
    assert failed_thread is not None
    failed_thread.join(timeout=2.0)
    assert not failed_thread.is_alive()
    failed = session.snapshot()

    assert failed["state"] == "failed"
    assert failed["outcome_unknown"] is True
    assert failed["can_open"] is False
    failure = failed["failure"]
    assert isinstance(failure, dict)
    assert failure["publish_attempts"] == 0
    assert failure["modeling_command_attempted"] is None
    assert failure["diagnostic_snapshot_unavailable"] == ["dialogue"]
    assert failure["diagnostic_evidence_unavailable"] == ["dialogue.modeling_command_attempted"]
    assert failure["safe_to_retry"] is False


def test_pretransport_authority_failure_remains_safe_after_worker_retirement() -> None:
    class FailingAuthorityLoader(FakeAuthorityLoader):
        def load(self) -> ApplicationControlAuthority:
            self.calls += 1
            raise RuntimeError("authority unavailable before transport creation")

    transport_factory_calls = 0

    def transport_factory(host: str) -> FakeTransport:
        nonlocal transport_factory_calls
        transport_factory_calls += 1
        return FakeTransport(host)

    session = InteractiveApplicationControlSession(
        FailingAuthorityLoader(),
        transport_factory=transport_factory,  # type: ignore[arg-type]
    )

    session.open(
        host="192.168.1.20",
        timezone_name="Europe/Moscow",
        confirmation=_confirmation(),
    )
    failed_thread = session._thread  # noqa: SLF001
    assert failed_thread is not None
    failed_thread.join(timeout=2.0)
    assert not failed_thread.is_alive()
    failed = session.snapshot()

    assert transport_factory_calls == 0
    assert failed["state"] == "failed"
    assert failed["outcome_unknown"] is False
    assert failed["can_open"] is True
    failure = failed["failure"]
    assert isinstance(failure, dict)
    assert failure["publish_attempts"] is None
    assert failure["modeling_command_attempted"] is False
    assert failure["diagnostic_snapshot_unavailable"] == []
    assert failure["diagnostic_evidence_unavailable"] == []
    assert failure["safe_to_retry"] is True