feat(k1): add physical acceptance transport

2026-07-18 15:17:46 +03:00
parent a4accf0fb6
commit eddc09008e
22 changed files with 1574 additions and 33 deletions
@@ -36,6 +36,10 @@ from k1link.device_plugins.xgrids_k1.mqtt import (
    capture_mqtt,
 )
 from k1link.device_plugins.xgrids_k1.net.snapshot import snapshot
+from k1link.device_plugins.xgrids_k1.protocol.application_authority import (
+    ApplicationAuthorityLoadError,
+    MacOSKeychainApplicationAuthorityProvisioner,
+)
 from k1link.device_plugins.xgrids_k1.usb.snapshot import snapshot as usb_snapshot
 from k1link.macos_credentials import CredentialDialogError, prompt_wifi_credentials

@@ -49,10 +53,15 @@ ble_app = typer.Typer(help="Bluetooth LE discovery and metadata commands.", no_a
 net_app = typer.Typer(help="Read-only local network observation commands.", no_args_is_help=True)
 usb_app = typer.Typer(help="Read-only macOS USB metadata commands.", no_args_is_help=True)
 analyze_app = typer.Typer(help="Bounded offline evidence analysis commands.", no_args_is_help=True)
+authority_app = typer.Typer(
+    help="One-time local authority administration; never sends a K1 command.",
+    no_args_is_help=True,
+)
 app.add_typer(ble_app, name="ble")
 app.add_typer(net_app, name="net")
 app.add_typer(usb_app, name="usb")
 app.add_typer(analyze_app, name="analyze")
+app.add_typer(authority_app, name="authority")


 class ToolStatus(TypedDict):
@@ -101,6 +110,39 @@ def _command_output(args: list[str]) -> str | None:
    return result.stdout.strip()


+@authority_app.command("provision")
+def authority_provision(
+    confirm_reviewed_authority: Annotated[
+        bool,
+        typer.Option(
+            "--confirm-reviewed-authority",
+            help="Confirm this is the reviewed private LixelGO FW 3.0.2 authority.",
+        ),
+    ] = False,
+) -> None:
+    """Prompt in macOS Keychain without receiving or printing the authority."""
+
+    if not confirm_reviewed_authority:
+        console.print(
+            "[red]Authority provisioning not confirmed.[/red] "
+            "Add --confirm-reviewed-authority only for the reviewed private value."
+        )
+        raise typer.Exit(code=2)
+    console.print(
+        "macOS Keychain will prompt for the reviewed authority. Mission Core will not "
+        "receive it through argv, environment, a file, browser state or logs."
+    )
+    try:
+        source = MacOSKeychainApplicationAuthorityProvisioner().provision_interactively()
+    except ApplicationAuthorityLoadError as exc:
+        console.print(f"[red]Authority provisioning failed:[/red] {exc}")
+        raise typer.Exit(code=2) from exc
+    console.print(
+        f"[green]Keychain item validated.[/green] service={source.service!r}; "
+        f"account={source.account!r}. No K1 command was sent."
+    )
+
+
 def _default_route_interface() -> str | None:
    output = _command_output(["route", "-n", "get", "default"])
    if output is None:
@@ -0,0 +1,188 @@
+from __future__ import annotations
+
+import math
+import threading
+import time
+from collections.abc import Callable, Collection, Sequence
+from dataclasses import dataclass
+from typing import Literal, Protocol
+
+from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import (
+    LiveDeviceControlBinding,
+    ShadowApplicationBootstrapOrchestrator,
+)
+from k1link.device_plugins.xgrids_k1.protocol.application_mqtt import (
+    MODELING_RESPONSE_TOPIC,
+)
+from k1link.device_plugins.xgrids_k1.protocol.application_publish import (
+    OneShotPublishEnvelope,
+)
+from k1link.device_plugins.xgrids_k1.protocol.modeling_control import (
+    ModelingAction,
+    ModelingResponse,
+    correlate_modeling_response,
+)
+from k1link.device_plugins.xgrids_k1.protocol.modeling_safety import (
+    ShadowModelingCommand,
+)
+
+MIN_ACCEPTANCE_PERMIT_SECONDS = 15.0
+MAX_ACCEPTANCE_PERMIT_SECONDS = 120.0
+
+
+class ApplicationAcceptanceError(RuntimeError):
+    """The operator-present physical acceptance contract failed closed."""
+
+
+class ApplicationBatchExchange(Protocol):
+    def exchange_batch_once(
+        self,
+        envelopes: Sequence[OneShotPublishEnvelope],
+        *,
+        required_response_topics: Collection[str],
+    ) -> dict[str, bytes]: ...
+
+
+@dataclass(frozen=True, slots=True)
+class PhysicalAcceptanceChecklist:
+    action: ModelingAction
+    operator_present: Literal[True]
+    owner_controlled_device: Literal[True]
+    lixelgo_closed: Literal[True]
+    battery_storage_confirmed: Literal[True]
+    expected_physical_state_confirmed: Literal[True]
+
+    def __post_init__(self) -> None:
+        confirmations = (
+            self.operator_present,
+            self.owner_controlled_device,
+            self.lixelgo_closed,
+            self.battery_storage_confirmed,
+            self.expected_physical_state_confirmed,
+        )
+        if any(value is not True for value in confirmations):
+            raise ApplicationAcceptanceError(
+                "every physical acceptance confirmation must be explicitly true"
+            )
+        if type(self.action) is not ModelingAction:
+            raise ApplicationAcceptanceError("acceptance action must be an explicit ModelingAction")
+
+
+class PhysicalAcceptancePermit:
+    """Short, single-action capability that is consumed before MQTT publish."""
+
+    def __init__(
+        self,
+        checklist: PhysicalAcceptanceChecklist,
+        *,
+        ttl_seconds: float = 60.0,
+        monotonic: Callable[[], float] = time.monotonic,
+    ) -> None:
+        if not isinstance(ttl_seconds, (int, float)) or isinstance(ttl_seconds, bool):
+            raise ApplicationAcceptanceError("acceptance permit TTL must be numeric")
+        if not math.isfinite(ttl_seconds) or not (
+            MIN_ACCEPTANCE_PERMIT_SECONDS <= ttl_seconds <= MAX_ACCEPTANCE_PERMIT_SECONDS
+        ):
+            raise ApplicationAcceptanceError(
+                "acceptance permit TTL is outside the reviewed 15-120 second range"
+            )
+        self._lock = threading.Lock()
+        self._checklist = checklist
+        self._monotonic = monotonic
+        self._expires_at = monotonic() + float(ttl_seconds)
+        self._consumed = False
+
+    @property
+    def action(self) -> ModelingAction:
+        return self._checklist.action
+
+    def consume(self, action: ModelingAction) -> None:
+        with self._lock:
+            if self._consumed:
+                raise ApplicationAcceptanceError("physical acceptance permit was already consumed")
+            if self._monotonic() >= self._expires_at:
+                raise ApplicationAcceptanceError("physical acceptance permit expired")
+            if action is not self._checklist.action:
+                raise ApplicationAcceptanceError("physical acceptance permit action mismatch")
+            self._consumed = True
+
+    def snapshot(self) -> dict[str, object]:
+        with self._lock:
+            remaining = max(0.0, self._expires_at - self._monotonic())
+            return {
+                "mode": "operator-present-physical-acceptance",
+                "action": self._checklist.action.name.casefold(),
+                "consumed": self._consumed,
+                "remaining_seconds": round(remaining, 3),
+                "automatic_retry": False,
+            }
+
+
+class PhysicalAcceptanceDialogueExecutor:
+    """Drive the recovered barriers and exactly one permitted START or STOP."""
+
+    def __init__(
+        self,
+        transport: ApplicationBatchExchange,
+        permit: PhysicalAcceptancePermit,
+    ) -> None:
+        self._transport = transport
+        self._permit = permit
+        self._bootstrap_complete = False
+        self._command_complete = False
+
+    def run_bootstrap(
+        self,
+        orchestrator: ShadowApplicationBootstrapOrchestrator,
+    ) -> LiveDeviceControlBinding:
+        if self._bootstrap_complete or self._command_complete:
+            raise ApplicationAcceptanceError("physical acceptance bootstrap was already attempted")
+        if self._permit.action is not ModelingAction.START:
+            raise ApplicationAcceptanceError("bootstrap is admitted only by a START permit")
+
+        while not orchestrator.snapshot().bootstrap_complete:
+            batch = orchestrator.next_batch()
+            required_topics = {
+                request.response_topic for request in batch if request.response_required
+            }
+            responses = self._transport.exchange_batch_once(
+                [OneShotPublishEnvelope.from_bootstrap_request(request) for request in batch],
+                required_response_topics=required_topics,
+            )
+            for request in batch:
+                if request.response_required:
+                    orchestrator.accept_response(
+                        request.response_topic,
+                        responses[request.response_topic],
+                    )
+
+        binding = orchestrator.binding
+        if binding is None:
+            raise ApplicationAcceptanceError("bootstrap completed without a live device binding")
+        self._bootstrap_complete = True
+        return binding
+
+    def execute_modeling(self, command: ShadowModelingCommand) -> ModelingResponse:
+        if self._command_complete:
+            raise ApplicationAcceptanceError("physical acceptance command was already attempted")
+        if command.action is ModelingAction.START and not self._bootstrap_complete:
+            raise ApplicationAcceptanceError("START requires the complete response-gated bootstrap")
+        self._permit.consume(command.action)
+        self._command_complete = True
+        responses = self._transport.exchange_batch_once(
+            [OneShotPublishEnvelope.from_modeling_command(command)],
+            required_response_topics={MODELING_RESPONSE_TOPIC},
+        )
+        return correlate_modeling_response(
+            responses[MODELING_RESPONSE_TOPIC],
+            command.command,
+        )
+
+    def snapshot(self) -> dict[str, object]:
+        return {
+            "mode": "physical-acceptance-only",
+            "bootstrap_complete": self._bootstrap_complete,
+            "command_complete": self._command_complete,
+            "permit": self._permit.snapshot(),
+            "automatic_retry": False,
+        }
@@ -3,6 +3,7 @@ from __future__ import annotations
 import platform
 import shutil
 import subprocess
+import sys
 from dataclasses import dataclass
 from typing import Protocol

@@ -14,6 +15,7 @@ from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import (
 KEYCHAIN_SERVICE = "NODEDC Mission Core XGRIDS K1 OpenAPI"
 KEYCHAIN_ACCOUNT = "lixelgo-application-fw-3.0.2"
 KEYCHAIN_TIMEOUT_SECONDS = 5.0
+KEYCHAIN_INTERACTIVE_TIMEOUT_SECONDS = 300.0


 class ApplicationAuthorityLoadError(RuntimeError):
@@ -31,6 +33,16 @@ class CommandRunner(Protocol):
    ) -> subprocess.CompletedProcess[bytes]: ...


+class InteractiveCommandRunner(Protocol):
+    def __call__(
+        self,
+        args: list[str],
+        *,
+        check: bool,
+        timeout: float,
+    ) -> subprocess.CompletedProcess[bytes]: ...
+
+
@dataclass(frozen=True, slots=True)
 class ApplicationAuthoritySourceSnapshot:
    provider: str = "macos-keychain"
@@ -116,3 +128,63 @@ class MacOSKeychainApplicationAuthorityLoader:
        finally:
            for index in range(len(secret_buffer)):
                secret_buffer[index] = 0
+
+
+class MacOSKeychainApplicationAuthorityProvisioner:
+    """Create or replace the fixed item through Apple's own hidden prompt.
+
+    ``security add-generic-password`` receives ``-w`` as its final option and
+    prompts on the controlling TTY. The authority therefore never enters a
+    Python string, subprocess argv, environment variable, file, log or browser
+    request. Provisioning is an explicit one-time operator/admin action and is
+    intentionally unavailable without an interactive terminal.
+    """
+
+    def __init__(
+        self,
+        *,
+        runner: InteractiveCommandRunner = subprocess.run,
+        loader: MacOSKeychainApplicationAuthorityLoader | None = None,
+    ) -> None:
+        self._runner = runner
+        self._loader = loader or MacOSKeychainApplicationAuthorityLoader()
+
+    def provision_interactively(self) -> ApplicationAuthoritySourceSnapshot:
+        if platform.system() != "Darwin":
+            raise ApplicationAuthorityLoadError(
+                "application authority provisioning is supported only through macOS Keychain"
+            )
+        if not sys.stdin.isatty() or not sys.stdout.isatty():
+            raise ApplicationAuthorityLoadError(
+                "application authority provisioning requires an interactive terminal"
+            )
+        security = shutil.which("security")
+        if security != "/usr/bin/security":
+            raise ApplicationAuthorityLoadError("trusted macOS security binary is unavailable")
+
+        try:
+            completed = self._runner(
+                [
+                    security,
+                    "add-generic-password",
+                    "-U",
+                    "-s",
+                    KEYCHAIN_SERVICE,
+                    "-a",
+                    KEYCHAIN_ACCOUNT,
+                    "-w",
+                ],
+                check=False,
+                timeout=KEYCHAIN_INTERACTIVE_TIMEOUT_SECONDS,
+            )
+        except (OSError, subprocess.SubprocessError) as exc:
+            raise ApplicationAuthorityLoadError(
+                "macOS Keychain authority provisioning failed"
+            ) from exc
+        if completed.returncode != 0:
+            raise ApplicationAuthorityLoadError("macOS Keychain authority was not provisioned")
+
+        # Re-open through the production loader so a wrong value fails before
+        # any Mission Core control lease can be armed.
+        self._loader.load()
+        return self._loader.snapshot()
@@ -0,0 +1,587 @@
+from __future__ import annotations
+
+import math
+import secrets
+import threading
+import time
+from collections import deque
+from collections.abc import Callable, Collection, Sequence
+from dataclasses import dataclass
+
+import paho.mqtt.client as mqtt
+from paho.mqtt.enums import CallbackAPIVersion
+from paho.mqtt.properties import Properties
+from paho.mqtt.reasoncodes import ReasonCode
+
+from k1link.device_plugins.xgrids_k1.ble.wifi_provisioning import AP_FALLBACK_IPV4
+from k1link.device_plugins.xgrids_k1.mqtt import validate_private_ipv4
+from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import (
+    DEVICE_CONFIG_RESPONSE_TOPIC,
+    DEVICE_INFO_RESPONSE_TOPIC,
+    GET_CLOUD_CONFIG_RESPONSE_TOPIC,
+    GET_NTRIP_PROFILE_RESPONSE_TOPIC,
+    GET_RTK_ADVANCE_RESPONSE_TOPIC,
+    MODELING_STATUS_RESPONSE_TOPIC,
+)
+from k1link.device_plugins.xgrids_k1.protocol.application_publish import (
+    OneShotPublishEnvelope,
+)
+from k1link.device_plugins.xgrids_k1.protocol.modeling_safety import (
+    MODELING_REQUEST_TOPIC,
+)
+
+MODELING_RESPONSE_TOPIC = "lixel/application/response/modeling"
+CONTROL_KEEPALIVE_SECONDS = 60
+CONTROL_CONNECT_TIMEOUT_SECONDS = 10.0
+CONTROL_EXCHANGE_TIMEOUT_SECONDS = 5.0
+CONTROL_LOOP_INTERVAL_SECONDS = 0.05
+MAX_CONTROL_RESPONSE_BYTES = 64 * 1024
+
+# The retained LixelGO control connection issued these three SUBSCRIBE packets
+# in this exact order. Point-cloud subscriptions belonged to a separate client.
+CONTROL_SUBSCRIPTION_GROUPS: tuple[tuple[tuple[str, int], ...], ...] = (
+    (
+        ("RealtimePath", 0),
+        ("RtkLioFusion", 0),
+        ("RtkPoseFusion", 0),
+        ("PrePathArray", 0),
+        ("ScanStatus", 0),
+        ("lixel/application/report/lio_pose", 0),
+        ("lixel/application/report/device_status", 0),
+        ("lixel/application/report/system_error", 0),
+        (DEVICE_INFO_RESPONSE_TOPIC, 0),
+    ),
+    (
+        ("SystemStatus", 1),
+        ("DeviceStatus", 1),
+        ("StatisticalData", 1),
+        ("lixel/application/report/modeling", 1),
+        ("lixel/application/report/upgrade", 1),
+    ),
+    (
+        ("AbnormalStatus", 2),
+        ("ActionApp2DeviceResponse", 2),
+        ("ActionAppSyncTimeResponse", 2),
+        ("DeviceInfoResponse", 2),
+        ("DeviceActivateResponse", 2),
+        ("CumulativeResponse", 2),
+        ("DebugStatusResponse", 2),
+        ("ModelingInfoResponse", 2),
+        ("ControlPointResponse", 2),
+        ("GetNtripConfigResponse", 2),
+        ("SetNtripConfigResponse", 2),
+        ("ProjectOperationResponse", 2),
+        ("PotreeConvertResponse", 2),
+        ("SetProjectNoteResponse", 2),
+        ("GetAppConfigResponse", 2),
+        ("SetAppConfigResponse", 2),
+        ("GetMqttConfigResponse", 2),
+        ("SetMqttConfigResponse", 2),
+        ("lixel/application/report/heartbeat", 2),
+        ("lixel/calibration/response/lidar", 2),
+        ("lixel/application/response/cameraConfig", 2),
+        ("lixel/application/response/service", 2),
+        ("lixel/application/report/pgo_progress", 2),
+        ("lixel/application/response/rtk/set_satellite_system", 2),
+        ("lixel/application/report/modeling/status", 2),
+        ("lixel/application/response/project_file", 2),
+        (GET_CLOUD_CONFIG_RESPONSE_TOPIC, 2),
+        ("lixel/application/response/set_cloud_server_config", 2),
+        (GET_NTRIP_PROFILE_RESPONSE_TOPIC, 2),
+        ("lixel/application/response/set_ntrip_profile", 2),
+        (GET_RTK_ADVANCE_RESPONSE_TOPIC, 2),
+        ("lixel/application/response/set_rtk_advance", 2),
+        (MODELING_STATUS_RESPONSE_TOPIC, 2),
+        ("lixel/application/response/control_point", 2),
+        ("lixel/application/response/service_control", 2),
+        ("lixel/application/response/project_list", 2),
+        ("lixel/application/response/device_control", 2),
+        (DEVICE_CONFIG_RESPONSE_TOPIC, 2),
+        (MODELING_RESPONSE_TOPIC, 2),
+        ("lixel/application/report/control_point", 2),
+        ("lixel/application/response/measure_point", 2),
+        ("lixel/application/response/upgrade", 2),
+    ),
+)
+
+APPLICATION_RESPONSE_TOPICS = frozenset(
+    {
+        DEVICE_INFO_RESPONSE_TOPIC,
+        MODELING_STATUS_RESPONSE_TOPIC,
+        GET_RTK_ADVANCE_RESPONSE_TOPIC,
+        DEVICE_CONFIG_RESPONSE_TOPIC,
+        GET_NTRIP_PROFILE_RESPONSE_TOPIC,
+        GET_CLOUD_CONFIG_RESPONSE_TOPIC,
+        MODELING_RESPONSE_TOPIC,
+    }
+)
+
+CONTROL_REPORT_TOPICS = frozenset(
+    topic
+    for group in CONTROL_SUBSCRIPTION_GROUPS
+    for topic, _qos in group
+    if "Response" not in topic and "/response/" not in topic
+)
+
+APPLICATION_REQUEST_ALLOWLIST = frozenset(
+    {
+        "lixel/application/request/device_info",
+        "lixel/application/request/modeling_status",
+        "lixel/application/request/get_rtk_advance",
+        "lixel/application/request/device_config",
+        "lixel/application/request/get_ntrip_profile",
+        "lixel/application/request/get_cloud_server_config",
+        MODELING_REQUEST_TOPIC,
+    }
+)
+
+
+class ApplicationMqttTransportError(RuntimeError):
+    """The reviewed control transport failed before a request was attempted."""
+
+
+class ApplicationCommandOutcomeUnknown(RuntimeError):
+    """A request may have reached the K1 and must never be retried automatically."""
+
+
+@dataclass(frozen=True, slots=True)
+class ApplicationMqttTransportSnapshot:
+    state: str
+    connect_attempts: int
+    subscribe_attempts: int
+    publish_attempts: int
+    qos2_completions: int
+    correlated_responses: int
+    ignored_known_responses: int
+    operation_keys_consumed: int
+    clean_session: bool = False
+    keepalive_seconds: int = CONTROL_KEEPALIVE_SECONDS
+    automatic_reconnect: bool = False
+    automatic_retry: bool = False
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "mode": "physical-acceptance-only",
+            "state": self.state,
+            "connect_attempts": self.connect_attempts,
+            "subscribe_attempts": self.subscribe_attempts,
+            "publish_attempts": self.publish_attempts,
+            "qos2_completions": self.qos2_completions,
+            "correlated_responses": self.correlated_responses,
+            "ignored_known_responses": self.ignored_known_responses,
+            "operation_keys_consumed": self.operation_keys_consumed,
+            "clean_session": self.clean_session,
+            "keepalive_seconds": self.keepalive_seconds,
+            "automatic_reconnect": self.automatic_reconnect,
+            "automatic_retry": self.automatic_retry,
+        }
+
+
+class ReviewedApplicationMqttTransport:
+    """Exact-profile MQTT exchange for an operator-present physical acceptance.
+
+    This type is deliberately not installed in the facade or plugin runtime.
+    It performs one connection attempt, never reconnects, consumes every
+    operation key before calling ``publish``, and permanently poisons itself
+    after any unknown post-publish outcome.
+    """
+
+    def __init__(
+        self,
+        host: str,
+        *,
+        port: int = 1883,
+        connect_timeout_seconds: float = CONTROL_CONNECT_TIMEOUT_SECONDS,
+        exchange_timeout_seconds: float = CONTROL_EXCHANGE_TIMEOUT_SECONDS,
+        client_factory: Callable[[], mqtt.Client] | None = None,
+        monotonic: Callable[[], float] = time.monotonic,
+    ) -> None:
+        self._target_ipv4 = validate_private_ipv4(host)
+        if self._target_ipv4 == AP_FALLBACK_IPV4:
+            raise ValueError("K1 access-point fallback address is not a direct-LAN control target")
+        if not 1 <= port <= 65535:
+            raise ValueError("port must be between 1 and 65535")
+        for name, value in (
+            ("connect_timeout_seconds", connect_timeout_seconds),
+            ("exchange_timeout_seconds", exchange_timeout_seconds),
+        ):
+            if not math.isfinite(value) or value <= 0:
+                raise ValueError(f"{name} must be finite and greater than zero")
+        self._port = port
+        self._connect_timeout_seconds = connect_timeout_seconds
+        self._exchange_timeout_seconds = exchange_timeout_seconds
+        self._monotonic = monotonic
+        self._client_factory = client_factory
+        self._client: mqtt.Client | None = None
+        self._lock = threading.Lock()
+        self._state = "new"
+        self._connected = False
+        self._subscribed = False
+        self._closing = False
+        self._subscription_mid: int | None = None
+        self._subscription_group_index = 0
+        self._completed_publish_mids: set[int] = set()
+        self._messages: deque[tuple[str, bytes]] = deque()
+        self._callback_error: str | None = None
+        self._consumed_operation_keys: set[str] = set()
+        self._connect_attempts = 0
+        self._subscribe_attempts = 0
+        self._publish_attempts = 0
+        self._qos2_completions = 0
+        self._correlated_responses = 0
+        self._ignored_known_responses = 0
+
+    def open(self) -> ApplicationMqttTransportSnapshot:
+        with self._lock:
+            if self._state != "new":
+                raise ApplicationMqttTransportError("control transport can be opened only once")
+            self._state = "connecting"
+            self._connect_attempts = 1
+        client = self._new_client()
+        self._install_callbacks(client)
+        self._client = client
+        try:
+            result = client.connect(
+                self._target_ipv4,
+                port=self._port,
+                keepalive=CONTROL_KEEPALIVE_SECONDS,
+            )
+        except (OSError, RuntimeError, ValueError) as exc:
+            self._fail_before_publish("control MQTT connect call failed", exc)
+        if result != mqtt.MQTT_ERR_SUCCESS:
+            self._fail_before_publish("control MQTT connect call was rejected")
+
+        deadline = self._monotonic() + self._connect_timeout_seconds
+        self._drive_until(lambda: self._subscribed, deadline, post_publish=False)
+        with self._lock:
+            self._state = "ready"
+        return self.snapshot()
+
+    def exchange_batch_once(
+        self,
+        envelopes: Sequence[OneShotPublishEnvelope],
+        *,
+        required_response_topics: Collection[str],
+    ) -> dict[str, bytes]:
+        batch = tuple(envelopes)
+        required = frozenset(required_response_topics)
+        if not batch:
+            raise ValueError("control exchange batch must not be empty")
+        if not required or not required <= APPLICATION_RESPONSE_TOPICS:
+            raise ValueError("required response topics exceed the reviewed allowlist")
+        operation_keys = tuple(envelope.operation_key for envelope in batch)
+        if len(set(operation_keys)) != len(operation_keys):
+            raise ValueError("control exchange batch repeats an operation key")
+        if any(envelope.topic not in APPLICATION_REQUEST_ALLOWLIST for envelope in batch):
+            raise ValueError("control exchange batch exceeds the reviewed request allowlist")
+        allowed_non_barrier_responses = (
+            frozenset({MODELING_STATUS_RESPONSE_TOPIC})
+            if any(
+                envelope.topic == "lixel/application/request/modeling_status" for envelope in batch
+            )
+            else frozenset()
+        )
+
+        with self._lock:
+            if self._consumed_operation_keys.intersection(operation_keys):
+                raise ApplicationCommandOutcomeUnknown(
+                    "control operation key was already consumed; retry is forbidden"
+                )
+            if self._state != "ready" or not self._connected or not self._subscribed:
+                raise ApplicationMqttTransportError("control transport is not ready")
+            stale_response = bool(self._messages)
+            if stale_response:
+                self._poison_locked("stale response preceded the request batch")
+            self._consumed_operation_keys.update(operation_keys)
+        if stale_response:
+            self.close()
+            raise ApplicationCommandOutcomeUnknown(
+                "stale control response makes the next command outcome ambiguous"
+            )
+
+        client = self._require_client()
+        publish_mids: set[int] = set()
+        for envelope in batch:
+            with self._lock:
+                self._publish_attempts += 1
+            try:
+                info = client.publish(
+                    envelope.topic,
+                    payload=envelope.payload,
+                    qos=envelope.qos,
+                    retain=envelope.retain,
+                )
+            except (OSError, RuntimeError, ValueError) as exc:
+                self._fail_after_publish("control MQTT publish call failed", exc)
+            if info.rc != mqtt.MQTT_ERR_SUCCESS or info.mid is None:
+                self._fail_after_publish("control MQTT publish call returned an unsafe result")
+            if info.mid in publish_mids:
+                self._fail_after_publish("control MQTT reused a packet identifier")
+            publish_mids.add(info.mid)
+
+        responses: dict[str, bytes] = {}
+        deadline = self._monotonic() + self._exchange_timeout_seconds
+
+        def complete() -> bool:
+            self._drain_responses(required, allowed_non_barrier_responses, responses)
+            with self._lock:
+                return publish_mids <= self._completed_publish_mids and required <= responses.keys()
+
+        self._drive_until(complete, deadline, post_publish=True)
+        self._drain_responses(required, allowed_non_barrier_responses, responses)
+        with self._lock:
+            self._correlated_responses += len(responses)
+        return responses
+
+    def close(self) -> None:
+        with self._lock:
+            if self._state == "closed":
+                return
+            self._closing = True
+            client = self._client
+        if client is not None:
+            try:
+                if self._subscribed:
+                    client.unsubscribe(
+                        [topic for group in CONTROL_SUBSCRIPTION_GROUPS for topic, _qos in group]
+                    )
+                client.disconnect()
+            except (OSError, RuntimeError, ValueError):
+                pass
+        with self._lock:
+            self._connected = False
+            self._subscribed = False
+            self._messages.clear()
+            if self._state not in {"poisoned", "failed"}:
+                self._state = "closed"
+
+    def snapshot(self) -> ApplicationMqttTransportSnapshot:
+        with self._lock:
+            return ApplicationMqttTransportSnapshot(
+                state=self._state,
+                connect_attempts=self._connect_attempts,
+                subscribe_attempts=self._subscribe_attempts,
+                publish_attempts=self._publish_attempts,
+                qos2_completions=self._qos2_completions,
+                correlated_responses=self._correlated_responses,
+                ignored_known_responses=self._ignored_known_responses,
+                operation_keys_consumed=len(self._consumed_operation_keys),
+            )
+
+    def _new_client(self) -> mqtt.Client:
+        if self._client_factory is not None:
+            return self._client_factory()
+        client_id = f"mck1-{secrets.token_hex(7)}"
+        return mqtt.Client(
+            callback_api_version=CallbackAPIVersion.VERSION2,
+            client_id=client_id,
+            clean_session=False,
+            protocol=mqtt.MQTTv311,
+            reconnect_on_failure=False,
+        )
+
+    def _install_callbacks(self, client: mqtt.Client) -> None:
+        def on_connect(
+            callback_client: mqtt.Client,
+            _userdata: object,
+            _flags: mqtt.ConnectFlags,
+            reason_code: ReasonCode,
+            _properties: Properties | None,
+        ) -> None:
+            if reason_code.is_failure:
+                self._set_callback_error("control MQTT broker rejected connection")
+                return
+            with self._lock:
+                self._connected = True
+            self._subscribe_next_group(callback_client)
+
+        def on_subscribe(
+            _callback_client: mqtt.Client,
+            _userdata: object,
+            mid: int,
+            reason_codes: list[ReasonCode],
+            _properties: Properties | None,
+        ) -> None:
+            with self._lock:
+                expected_mid = self._subscription_mid
+                group_index = self._subscription_group_index
+            if group_index >= len(CONTROL_SUBSCRIPTION_GROUPS):
+                self._set_callback_error("control MQTT received a surplus SUBACK")
+                return
+            expected_group = CONTROL_SUBSCRIPTION_GROUPS[group_index]
+            if mid != expected_mid or len(reason_codes) != len(expected_group):
+                self._set_callback_error("control MQTT received an unexpected SUBACK")
+                return
+            if any(reason_code.is_failure for reason_code in reason_codes):
+                self._set_callback_error("control MQTT broker rejected a response subscription")
+                return
+            with self._lock:
+                self._subscription_group_index += 1
+                complete = self._subscription_group_index == len(CONTROL_SUBSCRIPTION_GROUPS)
+                self._subscribed = complete
+            if not complete:
+                self._subscribe_next_group(_callback_client)
+
+        def on_publish(
+            _callback_client: mqtt.Client,
+            _userdata: object,
+            mid: int,
+            reason_code: ReasonCode,
+            _properties: Properties | None,
+        ) -> None:
+            if reason_code.is_failure:
+                self._set_callback_error("control MQTT QoS2 transaction failed")
+                return
+            with self._lock:
+                self._completed_publish_mids.add(mid)
+                self._qos2_completions += 1
+
+        def on_message(
+            _callback_client: mqtt.Client,
+            _userdata: object,
+            message: mqtt.MQTTMessage,
+        ) -> None:
+            allowed = {topic for group in CONTROL_SUBSCRIPTION_GROUPS for topic, _qos in group}
+            if message.topic not in allowed:
+                self._set_callback_error("control MQTT received an unreviewed subscribed topic")
+                return
+            payload = bytes(message.payload)
+            if len(payload) > MAX_CONTROL_RESPONSE_BYTES:
+                self._set_callback_error("control MQTT response exceeds the reviewed bound")
+                return
+            with self._lock:
+                if message.topic in CONTROL_REPORT_TOPICS:
+                    self._ignored_known_responses += 1
+                else:
+                    self._messages.append((message.topic, payload))
+
+        def on_disconnect(
+            _callback_client: mqtt.Client,
+            _userdata: object,
+            _flags: mqtt.DisconnectFlags,
+            reason_code: ReasonCode,
+            _properties: Properties | None,
+        ) -> None:
+            with self._lock:
+                expected = self._closing
+                self._connected = False
+            if not expected or reason_code.is_failure:
+                self._set_callback_error("control MQTT connection ended unexpectedly")
+
+        client.on_connect = on_connect
+        client.on_subscribe = on_subscribe
+        client.on_publish = on_publish
+        client.on_message = on_message
+        client.on_disconnect = on_disconnect
+
+    def _subscribe_next_group(self, client: mqtt.Client) -> None:
+        with self._lock:
+            group_index = self._subscription_group_index
+            if group_index < len(CONTROL_SUBSCRIPTION_GROUPS):
+                group = CONTROL_SUBSCRIPTION_GROUPS[group_index]
+                self._subscribe_attempts += 1
+            else:
+                group = None
+        if group is None:
+            self._set_callback_error("control MQTT subscription sequence overflowed")
+            return
+        try:
+            result, mid = client.subscribe(list(group))
+        except (OSError, RuntimeError, ValueError):
+            self._set_callback_error("control MQTT response subscription failed")
+            return
+        if result != mqtt.MQTT_ERR_SUCCESS or mid is None:
+            self._set_callback_error("control MQTT response subscription was rejected")
+            return
+        with self._lock:
+            self._subscription_mid = mid
+
+    def _drive_until(
+        self,
+        predicate: Callable[[], bool],
+        deadline: float,
+        *,
+        post_publish: bool,
+    ) -> None:
+        client = self._require_client()
+        while not predicate():
+            with self._lock:
+                callback_error = self._callback_error
+            if callback_error is not None:
+                if post_publish:
+                    self._fail_after_publish(callback_error)
+                self._fail_before_publish(callback_error)
+            if self._monotonic() >= deadline:
+                if post_publish:
+                    self._fail_after_publish("control MQTT response barrier timed out")
+                self._fail_before_publish("control MQTT connection/subscription timed out")
+            try:
+                result = client.loop(timeout=CONTROL_LOOP_INTERVAL_SECONDS)
+            except (OSError, RuntimeError, ValueError) as exc:
+                if post_publish:
+                    self._fail_after_publish("control MQTT network loop failed", exc)
+                self._fail_before_publish("control MQTT network loop failed", exc)
+            if result != mqtt.MQTT_ERR_SUCCESS:
+                if post_publish:
+                    self._fail_after_publish("control MQTT network loop returned an error")
+                self._fail_before_publish("control MQTT network loop returned an error")
+
+    def _drain_responses(
+        self,
+        required: frozenset[str],
+        allowed_non_barrier: frozenset[str],
+        responses: dict[str, bytes],
+    ) -> None:
+        ambiguous_message: str | None = None
+        with self._lock:
+            while self._messages:
+                topic, payload = self._messages.popleft()
+                if topic not in required:
+                    if topic in allowed_non_barrier:
+                        self._ignored_known_responses += 1
+                        continue
+                    self._poison_locked("unexpected response made correlation ambiguous")
+                    ambiguous_message = "unexpected control response made command outcome ambiguous"
+                    break
+                if topic in responses:
+                    self._poison_locked("duplicate required response made correlation ambiguous")
+                    ambiguous_message = "duplicate control response made command outcome ambiguous"
+                    break
+                responses[topic] = payload
+        if ambiguous_message is not None:
+            self.close()
+            raise ApplicationCommandOutcomeUnknown(ambiguous_message)
+
+    def _set_callback_error(self, message: str) -> None:
+        with self._lock:
+            if self._callback_error is None:
+                self._callback_error = message
+
+    def _fail_before_publish(self, message: str, cause: BaseException | None = None) -> None:
+        with self._lock:
+            self._state = "failed"
+        self.close()
+        error = ApplicationMqttTransportError(message)
+        if cause is not None:
+            raise error from cause
+        raise error
+
+    def _fail_after_publish(self, message: str, cause: BaseException | None = None) -> None:
+        with self._lock:
+            self._poison_locked(message)
+        self.close()
+        error = ApplicationCommandOutcomeUnknown(
+            f"{message}; automatic retry is forbidden until physical/status reconciliation"
+        )
+        if cause is not None:
+            raise error from cause
+        raise error
+
+    def _poison_locked(self, _reason: str) -> None:
+        self._state = "poisoned"
+
+    def _require_client(self) -> mqtt.Client:
+        client = self._client
+        if client is None:
+            raise ApplicationMqttTransportError("control MQTT client is not installed")
+        return client
@@ -45,6 +45,7 @@ class UninstalledApplicationPublishSink:

@dataclass(frozen=True, slots=True)
 class OneShotPublishEnvelope:
+    operation_key: str
    topic: str
    payload: bytes = field(repr=False)
    payload_sha256: str
@@ -59,6 +60,7 @@ class OneShotPublishEnvelope:
        request: EncodedApplicationRequest,
    ) -> OneShotPublishEnvelope:
        return cls(
+            operation_key=f"bootstrap:{request.ordinal}:{request.message_type}",
            topic=request.topic,
            payload=request.payload,
            payload_sha256=request.payload_sha256,
@@ -73,6 +75,7 @@ class OneShotPublishEnvelope:
        command: ShadowModelingCommand,
    ) -> OneShotPublishEnvelope:
        return cls(
+            operation_key=f"modeling:{command.action.name.casefold()}",
            topic=command.topic,
            payload=command.command.payload,
            payload_sha256=command.payload_sha256,
@@ -82,6 +85,10 @@ class OneShotPublishEnvelope:
        )

    def __post_init__(self) -> None:
+        if not self.operation_key or any(
+            ord(character) <= 0x20 or ord(character) > 0x7E for character in self.operation_key
+        ):
+            raise ValueError("publish envelope operation key is invalid")
        if self.qos != 2 or self.retain or self.automatic_retry:
            raise ValueError("publish envelope differs from the reviewed one-shot contract")
        if self.payload_bytes != len(self.payload):
@@ -91,6 +98,7 @@ class OneShotPublishEnvelope:

    def as_dict(self) -> dict[str, object]:
        return {
+            "operation_key": self.operation_key,
            "topic": self.topic,
            "payload_sha256": self.payload_sha256,
            "payload_bytes": self.payload_bytes,