feat(k1): add device-bound shadow control gate

docs/01_IMPLEMENTATION_PLAN.md

@@ -3,7 +3,7 @@
 This plan supersedes the app-dependent experiment order in the reference Bible.
 Each gate produces evidence and an explicit GO, PAUSE or BLOCKED result.
 
-## Current checkpoint — 2026-07-17
+## Current checkpoint — 2026-07-18
 
 | Stage | Result |
 | --- | --- |
@@ -21,6 +21,7 @@ Each gate produces evidence and an explicit GO, PAUSE or BLOCKED result.
 | Stage 7 observation archive | GO (point/pose/telemetry contract) — durable catalog, recovery, capture-clock-bounded RRD preparation, archived metric time series, saved-session timeline and atomic playback are implemented |
 | Stage 7 recorded cameras | GO (contract), acceptance pending — acquisition-owned fMP4 archive and player are implemented/tested; one real archived K1 camera plus point-cloud session has not passed playback yet |
 | Plugin isolation | GO (laboratory control plane) — vendor backend/frontend and optional scene controls are plugin-owned; manifest/runtime descriptor parity, versioned handshake, lifecycle health and transport correlation fail closed while execution remains in-process |
+| K1 application control | PAUSE (shadow gate implemented) — the exact captured codec is byte-matched, live status binds vendor identity plus serial, and START/STOP can be planned without publish authority; device-bound OpenAPI enrollment, one-shot publisher, physical save acceptance and recovery policy remain closed gates |
 | Stage 8 product storage | PAUSE — retention, replication, encryption, capacity monitoring and long-run browser/WASM stress remain deployment gates |
 
 USB project copying remains optional ground truth rather than a blocker for the
@@ -30,9 +31,13 @@ encoder, response correlator and device-status state machine now exist without a
 publisher. Header construction is known: an explicit ASCII device ID and ASCII
 OpenAPI key are required, while the session ID is derived exactly as
 `${device_id}:ModelingRequest`. MQTT control publishing remains deliberately
-disabled because legitimate OpenAPI credential provenance/provisioning and a
+disabled. A live-only shadow safety layer now observes the device-reported
-durable post-stop save gate are not proven. The physical button remains the
+vendor identity and serial, rejects identity drift or malformed status, and can
-known-safe fallback.
+form a non-executable START only from unbound `READY` or STOP only from a bound
+`SCANNING` state. Its output contains a digest and wire metadata, never the
+device identity, serial, OpenAPI value or payload. Legitimate per-device OpenAPI
+provisioning, an authorized one-shot publisher and a durable post-stop save gate
+are not proven. The physical button remains the known-safe fallback.
 
 The Stage 6 live path uses a bounded raw-first bridge: loss in the visualization
 queue cannot discard MQTT evidence. The queue holds four preview messages;

docs/05_K1_MQTT_STREAM_PROFILE.md

@@ -157,13 +157,37 @@ correlated them with device success responses and physical start/stop of the
 high-rate streams. This promotes the mapping from a static hint to descriptive
 wire evidence; it does not authorize replay or publishing.
 
+An offline bounded audit of the retained owner-controlled captures additionally
+proved that the repository encoder reproduces the observed START and STOP
+payloads byte-for-byte for this one K1 and exact firmware profile. It also
+separated four values which must never be substituted for one another:
+
+- Mission Core's provisional device UUID is local inventory identity and never
+  belongs in a vendor request;
+- the vendor device ID comes from live K1 status and occupies the request header;
+- the K1 serial is a separate live device-binding value;
+- the OpenAPI value is a private, device-bound command credential and is not
+  present in the status report.
+
+Consequently, support for another K1 cannot replay this scanner's captured
+payload or OpenAPI value. It requires a separate owner-authorized enrollment
+whose vendor device ID and serial are attested against live status under the
+same exact firmware/topology profile. Any mismatch, identity drift, malformed
+status, replay-only evidence or unexpected lifecycle state fails closed.
+
 The repository now contains an inert bounded encoder/response parser and a
-fail-closed device-status state machine for this exact profile. They have no
+fail-closed device-status state machine for this exact profile. A live-only
-MQTT publish dependency. Publishing remains deliberately disabled: the OpenAPI
+shadow planner accepts START only from `READY` with no project and STOP only
-credential's provenance and secure provisioning are unresolved, and the
+from `SCANNING` with a bound project. It emits only a non-executable digest and
-complete `ScanOver`/ready/stable-artifact sequence has not physically proved a
+wire metadata, uses QoS 2 with retain false, and declares automatic retry
-durable save result. The verified physical double-click remains the acquisition
+forbidden. It has no MQTT publish dependency and cannot send a command.
-control until a separate reviewed write gate closes.
+
+Publishing remains deliberately disabled: secure per-device OpenAPI enrollment
+is unresolved, and the complete stop/save sequence has not physically proved a
+durable vendor project. A correlated STOP success, stream quiescence, return to
+`READY`/steady-green hardware state and appearance of a reusable native project
+are separate gates. The verified physical double-click remains the acquisition
+control until a separately reviewed one-shot write gate closes.
 
 ## Decoder and raw-capture bounds
 

docs/adr/0012-device-bound-k1-command-authority.md

@@ -0,0 +1,94 @@
+# ADR 0012: device-bound K1 command authority
+
+- Status: accepted
+- Date: 2026-07-18
+- Extends: ADR 0004, ADR 0005, ADR 0009 and ADR 0010
+
+## Context
+
+Owner-controlled LixelGO traffic proves the K1 modeling START/STOP protobuf
+shape, MQTT topic, QoS and correlated success response for one K1 running the
+exact `xgrids.lixelkity-k1.fw-3.0.2.direct-lan.v1` profile. A bounded offline
+audit also proves that Mission Core's encoder reproduces both retained requests
+byte-for-byte. This is sufficient to describe the current device dialogue, but
+not to replay a captured request or authorize writes to any K1.
+
+The protocol exposes several identities with different authority. Mission
+Core's provisional device UUID is local inventory identity. The vendor device
+ID is reported by the live K1 and occupies the request header. The K1 serial is
+a separate device-binding value. The OpenAPI value is private command material
+which is absent from live status. LixelGO derives the command session relation
+from the vendor device ID and request type; it is not a reusable caller session.
+
+Native project durability is also not equivalent to command acceptance. STOP
+success, stream quiescence, return to READY/steady green and a reusable project
+on the scanner are separate observations.
+
+## Decision
+
+K1 application control is exact-profile, device-bound and fail-closed.
+
+1. Mission Core never sends its local inventory UUID, BLE identifier or a K1
+   serial in place of the vendor request identity.
+2. Each K1 requires an owner-authorized enrollment containing its vendor device
+   ID, serial and private OpenAPI value. Enrollment secrets are not committed,
+   placed in Ops, returned by APIs or written to logs.
+3. Before a command can be considered, live MQTT status must attest the same
+   vendor device ID and serial under the exact firmware/topology profile.
+   Replay evidence cannot establish current command authority.
+4. Identity drift, missing identity, malformed status, profile mismatch or an
+   unexpected lifecycle state blocks command planning.
+5. START retains the observed LixelGO parameters: record-and-calculate mode 2,
+   LCC scan mode 1, handheld mount mode 0 and the operator's validated project
+   name. STOP contains only the same bound header and action.
+6. MQTT command semantics remain QoS 2, retain false. Automatic retry is
+   forbidden: an unknown outcome must be reconciled from live status and
+   physical evidence before another command.
+7. No firmware, activation, account, update or vendor-cloud endpoint is part of
+   the control path. Local direct-LAN MQTT is the only reviewed transport.
+8. STOP completion has four gates: correlated command result, local stream
+   quiescence and evidence sealing, READY/steady-green device state, then native
+   project verification. Earlier gates must not claim the later ones.
+
+## Current implementation boundary
+
+`LiveModelingControlSafety` observes only live device-status messages. It binds
+vendor identity and serial, tracks lifecycle state and creates non-executable
+shadow START/STOP plans. Public state and plan output contain only booleans,
+counts, action, topic, QoS, retain flag, payload length and digest. The encoded
+payload and enrolled values remain private in memory.
+
+The shadow plan is blocked by both `vendor-writes-disabled` and
+`publisher-not-installed`. There is no MQTT publisher, enrollment endpoint or
+automatic retry path. The physical K1 control remains the production fallback.
+
+## Promotion gate
+
+A future publisher requires a separate review and an operator-present physical
+acceptance on the enrolled K1:
+
+1. confirm exact firmware/profile, steady-green state, battery and storage;
+2. close LixelGO and arm raw evidence plus the intended camera receiver;
+3. attest live READY identity against the device enrollment;
+4. review the shadow command metadata;
+5. send one START, without retry, and correlate its exact response;
+6. observe calibration and first point/pose/camera data;
+7. send one STOP, without retry, and correlate its exact response;
+8. seal local evidence while accepting the bounded stream tail;
+9. wait for READY and steady green before power-off or another scan;
+10. verify the native project independently through the vendor-supported
+    workflow.
+
+Any unknown response, identity change, activation state, fault, low battery or
+unexpected lifecycle transition pauses the test. It does not trigger a guessed
+recovery command.
+
+## Consequences
+
+Adding another K1 means creating another private enrollment and proving its
+exact compatibility profile; it never means reusing captured bytes from the
+first scanner. The design preserves native record-and-calculate behavior while
+keeping command authority inside the XGRIDS plugin and out of generic Mission
+Core. It intentionally postpones convenience automation until identity,
+credential provenance, one-shot transport behavior and durable save semantics
+have all passed physical acceptance.

plugins/xgrids-k1/frontend/src/api.ts

@@ -67,6 +67,21 @@ export interface XgridsCompatibilityState {
   camera_preview?: string | null;
 }
 
+export interface XgridsModelingControlSafety {
+  mode: "shadow-only";
+  status_reports_observed: number;
+  decode_errors: number;
+  vendor_identity_observed: boolean;
+  device_serial_observed: boolean;
+  identity_conflict: boolean;
+  session_state?: string | null;
+  project_bound: boolean;
+  ready_for_start_shadow: boolean;
+  ready_for_stop_shadow: boolean;
+  vendor_writes_enabled: false;
+  publisher_installed: false;
+}
+
 export interface XgridsAcquisition {
   schema_version?: string;
   acquisition_id: string;
@@ -189,6 +204,7 @@ export interface XgridsK1State {
   source_mode?: SourceMode | null;
   metrics?: XgridsK1Metrics;
   compatibility?: XgridsCompatibilityState | null;
+  modeling_control_safety?: XgridsModelingControlSafety | null;
   device_ref?: XgridsDeviceRef | null;
   device_session?: XgridsDeviceSession | null;
   acquisition?: XgridsAcquisition | null;

src/k1link/device_plugins/xgrids_k1/facade.py

@@ -46,12 +46,17 @@ from k1link.device_plugins.xgrids_k1.camera import (
 from k1link.device_plugins.xgrids_k1.mqtt import validate_private_ipv4
 from k1link.device_plugins.xgrids_k1.mqtt.capture import seal_capture_clock
 from k1link.device_plugins.xgrids_k1.protocol.modeling import observe_modeling_report
+from k1link.device_plugins.xgrids_k1.protocol.modeling_safety import (
+    LiveModelingControlSafety,
+)
 from k1link.device_plugins.xgrids_k1.protocol.normalizer import normalize_k1_message
+from k1link.device_plugins.xgrids_k1.viewer.messages import StreamMessage
 from k1link.device_plugins.xgrids_k1.viewer.runtime import (
     VisualizationRuntime,
     new_live_session_dir,
 )
 from k1link.sessions import ActiveSessionLease, resolve_missioncore_evidence_dir
+from k1link.viewer.metrics import BridgeMetrics
 from k1link.viewer.rerun_bridge import RerunSceneSettings
 from k1link.web.device_lifecycle import (
     TERMINAL_ACQUISITION_STATES,
@@ -278,9 +283,10 @@ class XgridsK1CompatibilityService:
         self._acquisition_session_lease: ActiveSessionLease | None = None
         # The host-owned visual runtime receives the vendor normalizer
         # explicitly. There is no implicit K1 decoder in the visual layer.
+        self._modeling_control_safety = LiveModelingControlSafety()
         self.runtime = VisualizationRuntime(
             normalizer=normalize_k1_message,
-            message_observer=observe_modeling_report,
+            message_observer=self._observe_runtime_message,
         )
         self.camera_preview = XgridsK1CameraGateway(
             self.repository_root,
@@ -379,6 +385,7 @@ class XgridsK1CompatibilityService:
                     else "unverified"
                 ),
             },
+            "modeling_control_safety": self._modeling_control_safety.snapshot().as_dict(),
             "device_ref": (
                 {
                     "device_id": device_id,
@@ -645,6 +652,15 @@ class XgridsK1CompatibilityService:
 
         return self.state()
 
+    def _observe_runtime_message(
+        self,
+        message: StreamMessage,
+        metrics: BridgeMetrics,
+    ) -> bool:
+        if self._modeling_control_safety.observe(message, metrics):
+            return True
+        return observe_modeling_report(message, metrics)
+
     def verify_connection(self) -> dict[str, Any]:
         """Validate the recorded endpoint only; no network packet is emitted."""
 
@@ -842,6 +858,7 @@ class XgridsK1CompatibilityService:
                     lease.release()
                     raise RuntimeError("evidence-сессия уже удерживается активным acquisition")
                 self._acquisition_session_lease = lease
+            self._modeling_control_safety.reset()
             self.runtime.start_live(
                 acquisition.target_host,
                 out_dir,

src/k1link/device_plugins/xgrids_k1/protocol/modeling_safety.py

@@ -0,0 +1,275 @@
+from __future__ import annotations
+
+import hashlib
+import threading
+from dataclasses import dataclass, field
+from typing import Literal
+
+from k1link.device_plugins.xgrids_k1.protocol.modeling_control import (
+    CommandHeaderIdentity,
+    DeviceStatusReport,
+    EncodedModelingCommand,
+    ModelingAction,
+    ModelingEncodeError,
+    MountType,
+    RecordMode,
+    ScanMode,
+    SessionState,
+    decode_device_status_report,
+    encode_modeling_start,
+    encode_modeling_stop,
+)
+from k1link.device_plugins.xgrids_k1.viewer.messages import StreamMessage
+from k1link.viewer.metrics import BridgeMetrics
+
+DEVICE_STATUS_TOPIC = "lixel/application/report/device_status"
+MODELING_REQUEST_TOPIC = "lixel/application/request/modeling"
+SHADOW_BLOCKERS = ("vendor-writes-disabled", "publisher-not-installed")
+CONTROL_ENROLLMENT_SOURCE = "owner-captured-device-bound"
+CONTROL_COMPATIBILITY_PROFILE_ID = "xgrids.lixelkity-k1.fw-3.0.2.direct-lan.v1"
+CONTROL_FIRMWARE_VERSION = "3.0.2"
+CONTROL_TOPOLOGY = "direct-lan"
+
+
+class ModelingControlSafetyError(RuntimeError):
+    """A live K1 could not be bound to reviewed device-specific control input."""
+
+
+@dataclass(frozen=True, slots=True)
+class DeviceBoundControlEnrollment:
+    """Private, device-specific values learned from owner-controlled evidence.
+
+    Mission Core's provisional device UUID, BLE transport UUID and K1 serial are
+    not interchangeable with the vendor header identity.  An enrollment binds
+    all three vendor inputs explicitly and keeps them out of repr/API output.
+    """
+
+    vendor_device_id: str = field(repr=False)
+    device_serial: str = field(repr=False)
+    openapi_key: str = field(repr=False)
+    source: Literal["owner-captured-device-bound"] = "owner-captured-device-bound"
+    compatibility_profile_id: Literal["xgrids.lixelkity-k1.fw-3.0.2.direct-lan.v1"] = (
+        "xgrids.lixelkity-k1.fw-3.0.2.direct-lan.v1"
+    )
+    firmware_version: Literal["3.0.2"] = "3.0.2"
+    topology: Literal["direct-lan"] = "direct-lan"
+
+    def __post_init__(self) -> None:
+        if (
+            self.source != CONTROL_ENROLLMENT_SOURCE
+            or self.compatibility_profile_id != CONTROL_COMPATIBILITY_PROFILE_ID
+            or self.firmware_version != CONTROL_FIRMWARE_VERSION
+            or self.topology != CONTROL_TOPOLOGY
+        ):
+            raise ModelingControlSafetyError(
+                "control enrollment does not match the exact reviewed K1 profile"
+            )
+        CommandHeaderIdentity(
+            device_id=self.vendor_device_id,
+            openapi_key=self.openapi_key,
+        )
+        _validate_device_serial(self.device_serial)
+
+    @property
+    def command_header(self) -> CommandHeaderIdentity:
+        return CommandHeaderIdentity(
+            device_id=self.vendor_device_id,
+            openapi_key=self.openapi_key,
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class ModelingControlSafetySnapshot:
+    status_reports_observed: int
+    decode_errors: int
+    vendor_identity_observed: bool
+    device_serial_observed: bool
+    identity_conflict: bool
+    session_state: str | None
+    project_bound: bool
+    ready_for_start_shadow: bool
+    ready_for_stop_shadow: bool
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "mode": "shadow-only",
+            "status_reports_observed": self.status_reports_observed,
+            "decode_errors": self.decode_errors,
+            "vendor_identity_observed": self.vendor_identity_observed,
+            "device_serial_observed": self.device_serial_observed,
+            "identity_conflict": self.identity_conflict,
+            "session_state": self.session_state,
+            "project_bound": self.project_bound,
+            "ready_for_start_shadow": self.ready_for_start_shadow,
+            "ready_for_stop_shadow": self.ready_for_stop_shadow,
+            "vendor_writes_enabled": False,
+            "publisher_installed": False,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class ShadowModelingCommand:
+    """A reviewed command plan with no transport or execution authority."""
+
+    action: ModelingAction
+    command: EncodedModelingCommand = field(repr=False)
+    payload_sha256: str
+    payload_bytes: int
+    topic: str = MODELING_REQUEST_TOPIC
+    qos: int = 2
+    retain: bool = False
+    executable: bool = False
+    blockers: tuple[str, ...] = SHADOW_BLOCKERS
+    retry_policy: Literal["never-automatic"] = "never-automatic"
+
+    @classmethod
+    def from_command(cls, command: EncodedModelingCommand) -> ShadowModelingCommand:
+        return cls(
+            action=command.action,
+            command=command,
+            payload_sha256=hashlib.sha256(command.payload).hexdigest(),
+            payload_bytes=len(command.payload),
+        )
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "mode": "shadow-only",
+            "action": self.action.name.casefold(),
+            "topic": self.topic,
+            "qos": self.qos,
+            "retain": self.retain,
+            "payload_sha256": self.payload_sha256,
+            "payload_bytes": self.payload_bytes,
+            "executable": self.executable,
+            "blockers": list(self.blockers),
+            "retry_policy": self.retry_policy,
+        }
+
+
+class LiveModelingControlSafety:
+    """Track only live status evidence and fail closed on identity drift."""
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._status_reports_observed = 0
+        self._decode_errors = 0
+        self._vendor_device_id: str | None = None
+        self._device_serial: str | None = None
+        self._identity_conflict = False
+        self._latest_report: DeviceStatusReport | None = None
+
+    def reset(self) -> None:
+        with self._lock:
+            self._status_reports_observed = 0
+            self._decode_errors = 0
+            self._vendor_device_id = None
+            self._device_serial = None
+            self._identity_conflict = False
+            self._latest_report = None
+
+    def observe(self, message: StreamMessage, _metrics: BridgeMetrics) -> bool:
+        if message.source != "live_mqtt" or message.topic != DEVICE_STATUS_TOPIC:
+            return False
+        try:
+            report = decode_device_status_report(message.payload)
+        except ValueError:
+            with self._lock:
+                self._decode_errors += 1
+            return True
+
+        header_device_id = report.header.device_id if report.header is not None else None
+        with self._lock:
+            self._status_reports_observed += 1
+            if header_device_id is None or report.device_sn is None:
+                self._identity_conflict = True
+            else:
+                if self._vendor_device_id is None:
+                    self._vendor_device_id = header_device_id
+                elif self._vendor_device_id != header_device_id:
+                    self._identity_conflict = True
+                if self._device_serial is None:
+                    self._device_serial = report.device_sn
+                elif self._device_serial != report.device_sn:
+                    self._identity_conflict = True
+            self._latest_report = report
+        return True
+
+    def snapshot(self) -> ModelingControlSafetySnapshot:
+        with self._lock:
+            report = self._latest_report
+            state = report.session_state if report is not None else None
+            project_bound = bool(report and report.project_id)
+            observed = self._vendor_device_id is not None and self._device_serial is not None
+            safe = observed and not self._identity_conflict and self._decode_errors == 0
+            return ModelingControlSafetySnapshot(
+                status_reports_observed=self._status_reports_observed,
+                decode_errors=self._decode_errors,
+                vendor_identity_observed=self._vendor_device_id is not None,
+                device_serial_observed=self._device_serial is not None,
+                identity_conflict=self._identity_conflict,
+                session_state=state.name.casefold() if state is not None else None,
+                project_bound=project_bound,
+                ready_for_start_shadow=safe and state is SessionState.READY and not project_bound,
+                ready_for_stop_shadow=safe and state is SessionState.SCANNING and project_bound,
+            )
+
+    def plan_start(
+        self,
+        enrollment: DeviceBoundControlEnrollment,
+        *,
+        project_name: str,
+    ) -> ShadowModelingCommand:
+        self._require_bound(enrollment, required_state=SessionState.READY, project_bound=False)
+        command = encode_modeling_start(
+            enrollment.command_header,
+            project_name=project_name,
+            record_mode=RecordMode.RECORD_AND_CALCULATE,
+            scan_mode=ScanMode.LCC,
+            mount_type=MountType.HANDHELD,
+        )
+        return ShadowModelingCommand.from_command(command)
+
+    def plan_stop(
+        self,
+        enrollment: DeviceBoundControlEnrollment,
+    ) -> ShadowModelingCommand:
+        self._require_bound(enrollment, required_state=SessionState.SCANNING, project_bound=True)
+        return ShadowModelingCommand.from_command(encode_modeling_stop(enrollment.command_header))
+
+    def _require_bound(
+        self,
+        enrollment: DeviceBoundControlEnrollment,
+        *,
+        required_state: SessionState,
+        project_bound: bool,
+    ) -> None:
+        with self._lock:
+            report = self._latest_report
+            if self._identity_conflict or self._decode_errors:
+                raise ModelingControlSafetyError("live K1 identity/status evidence is conflicted")
+            if report is None or self._vendor_device_id is None or self._device_serial is None:
+                raise ModelingControlSafetyError("live K1 identity/status evidence is missing")
+            if (
+                enrollment.vendor_device_id != self._vendor_device_id
+                or enrollment.device_serial != self._device_serial
+            ):
+                raise ModelingControlSafetyError(
+                    "device-bound control enrollment does not match the live K1"
+                )
+            if report.session_state is not required_state:
+                raise ModelingControlSafetyError(
+                    f"live K1 must be {required_state.name.casefold()} for this shadow plan"
+                )
+            if bool(report.project_id) is not project_bound:
+                raise ModelingControlSafetyError("live K1 project binding is not in the safe state")
+
+
+def _validate_device_serial(value: object) -> None:
+    if not isinstance(value, str) or not value:
+        raise ModelingEncodeError("device_serial must be a non-empty string")
+    try:
+        encoded = value.encode("ascii")
+    except UnicodeEncodeError as exc:
+        raise ModelingEncodeError("device_serial must use printable ASCII") from exc
+    if len(encoded) > 256 or any(byte <= 0x20 or byte > 0x7E for byte in encoded):
+        raise ModelingEncodeError("device_serial must use bounded printable ASCII without spaces")

tests/test_xgrids_modeling_safety.py

@@ -0,0 +1,175 @@
+from __future__ import annotations
+
+import pytest
+
+from k1link.device_plugins.xgrids_k1.protocol.modeling_control import (
+    MODELING_STATE_BASE,
+    ModelingAction,
+    SessionState,
+)
+from k1link.device_plugins.xgrids_k1.protocol.modeling_safety import (
+    DEVICE_STATUS_TOPIC,
+    DeviceBoundControlEnrollment,
+    LiveModelingControlSafety,
+    ModelingControlSafetyError,
+)
+from k1link.device_plugins.xgrids_k1.protocol.protobuf_wire import iter_fields
+from k1link.device_plugins.xgrids_k1.viewer.messages import StreamMessage
+from k1link.viewer.metrics import BridgeMetrics
+
+
+def _varint(value: int) -> bytes:
+    encoded = bytearray()
+    while value > 0x7F:
+        encoded.append((value & 0x7F) | 0x80)
+        value >>= 7
+    encoded.append(value)
+    return bytes(encoded)
+
+
+def _uint(number: int, value: int) -> bytes:
+    return _varint(number << 3) + _varint(value)
+
+
+def _bytes(number: int, value: bytes) -> bytes:
+    return _varint((number << 3) | 2) + _varint(len(value)) + value
+
+
+def _text(number: int, value: str) -> bytes:
+    return _bytes(number, value.encode())
+
+
+def _status_message(
+    state: SessionState,
+    *,
+    device_id: str = "vendor-device-001",
+    device_serial: str = "serial-001",
+    project_id: str | None = None,
+    source: str = "live_mqtt",
+) -> StreamMessage:
+    header = _text(4, device_id) + _text(5, "status-message-session")
+    payload = _bytes(1, header) + _uint(2, MODELING_STATE_BASE + state) + _text(3, device_serial)
+    if project_id is not None:
+        payload += _text(4, project_id)
+    return StreamMessage(
+        sequence=1,
+        topic=DEVICE_STATUS_TOPIC,
+        payload=payload,
+        received_at_epoch_ns=1,
+        received_monotonic_ns=1,
+        source=source,
+    )
+
+
+def _enrollment(**changes: str) -> DeviceBoundControlEnrollment:
+    values = {
+        "vendor_device_id": "vendor-device-001",
+        "device_serial": "serial-001",
+        "openapi_key": "device-bound-private-key",
+    }
+    values.update(changes)
+    return DeviceBoundControlEnrollment(**values)
+
+
+def test_live_status_identity_is_observed_without_accepting_replay() -> None:
+    safety = LiveModelingControlSafety()
+    metrics = BridgeMetrics()
+
+    assert not safety.observe(
+        _status_message(SessionState.READY, source="k1mqtt"),
+        metrics,
+    )
+    assert safety.snapshot().status_reports_observed == 0
+
+    assert safety.observe(_status_message(SessionState.READY), metrics)
+    snapshot = safety.snapshot()
+    assert snapshot.status_reports_observed == 1
+    assert snapshot.vendor_identity_observed
+    assert snapshot.device_serial_observed
+    assert snapshot.ready_for_start_shadow
+    assert not snapshot.ready_for_stop_shadow
+    assert "vendor-device" not in repr(snapshot)
+
+
+def test_start_shadow_plan_is_exact_bounded_and_never_executable() -> None:
+    safety = LiveModelingControlSafety()
+    safety.observe(_status_message(SessionState.READY), BridgeMetrics())
+
+    plan = safety.plan_start(_enrollment(), project_name="SAFE_PROJECT")
+    fields = {field.number: field for field in iter_fields(plan.command.payload, max_fields=16)}
+
+    assert plan.action is ModelingAction.START
+    assert fields[2].value == ModelingAction.START
+    assert fields[3].value == b"SAFE_PROJECT"
+    assert fields[4].value == 2
+    assert fields[5].value == 1
+    assert 6 not in fields
+    assert plan.qos == 2
+    assert plan.retain is False
+    assert plan.executable is False
+    assert plan.blockers == ("vendor-writes-disabled", "publisher-not-installed")
+    assert plan.retry_policy == "never-automatic"
+    assert "device-bound-private-key" not in repr(plan)
+    assert "device-bound-private-key" not in str(plan.as_dict())
+
+
+def test_stop_shadow_plan_requires_same_live_device_and_active_project() -> None:
+    safety = LiveModelingControlSafety()
+    safety.observe(
+        _status_message(SessionState.SCANNING, project_id="private-project-id"),
+        BridgeMetrics(),
+    )
+
+    plan = safety.plan_stop(_enrollment())
+    fields = list(iter_fields(plan.command.payload, max_fields=8))
+    assert plan.action is ModelingAction.STOP
+    assert [field.number for field in fields] == [1, 2]
+
+    with pytest.raises(ModelingControlSafetyError, match="does not match"):
+        safety.plan_stop(_enrollment(device_serial="another-serial"))
+
+
+def test_identity_drift_and_decode_failure_fail_closed() -> None:
+    safety = LiveModelingControlSafety()
+    metrics = BridgeMetrics()
+    safety.observe(_status_message(SessionState.READY), metrics)
+    safety.observe(
+        _status_message(SessionState.READY, device_id="different-vendor-device"),
+        metrics,
+    )
+    assert safety.snapshot().identity_conflict
+    with pytest.raises(ModelingControlSafetyError, match="conflicted"):
+        safety.plan_start(_enrollment(), project_name="SAFE_PROJECT")
+
+    safety.reset()
+    malformed = StreamMessage(
+        sequence=2,
+        topic=DEVICE_STATUS_TOPIC,
+        payload=b"\x10",
+        received_at_epoch_ns=2,
+        received_monotonic_ns=2,
+        source="live_mqtt",
+    )
+    assert safety.observe(malformed, metrics)
+    assert safety.snapshot().decode_errors == 1
+    with pytest.raises(ModelingControlSafetyError, match="conflicted"):
+        safety.plan_start(_enrollment(), project_name="SAFE_PROJECT")
+
+
+def test_shadow_plan_requires_exact_safe_lifecycle_state() -> None:
+    safety = LiveModelingControlSafety()
+    safety.observe(_status_message(SessionState.SCAN_STARTING), BridgeMetrics())
+    with pytest.raises(ModelingControlSafetyError, match="must be ready"):
+        safety.plan_start(_enrollment(), project_name="SAFE_PROJECT")
+    with pytest.raises(ModelingControlSafetyError, match="must be scanning"):
+        safety.plan_stop(_enrollment())
+
+
+def test_enrollment_rejects_runtime_profile_substitution() -> None:
+    with pytest.raises(ModelingControlSafetyError, match="exact reviewed K1 profile"):
+        DeviceBoundControlEnrollment(
+            vendor_device_id="vendor-device-001",
+            device_serial="serial-001",
+            openapi_key="device-bound-private-key",
+            firmware_version="3.0.3",  # type: ignore[arg-type]
+        )