diff --git a/docs/01_IMPLEMENTATION_PLAN.md b/docs/01_IMPLEMENTATION_PLAN.md index 0afa415..232e439 100644 --- a/docs/01_IMPLEMENTATION_PLAN.md +++ b/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 -durable post-stop save gate are not proven. The physical button remains the -known-safe fallback. +disabled. A live-only shadow safety layer now observes the device-reported +vendor identity and serial, rejects identity drift or malformed status, and can +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; diff --git a/docs/05_K1_MQTT_STREAM_PROFILE.md b/docs/05_K1_MQTT_STREAM_PROFILE.md index e755160..755fdcc 100644 --- a/docs/05_K1_MQTT_STREAM_PROFILE.md +++ b/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 -MQTT publish dependency. Publishing remains deliberately disabled: the OpenAPI -credential's provenance and secure provisioning are unresolved, and the -complete `ScanOver`/ready/stable-artifact sequence has not physically proved a -durable save result. The verified physical double-click remains the acquisition -control until a separate reviewed write gate closes. +fail-closed device-status state machine for this exact profile. A live-only +shadow planner accepts START only from `READY` with no project and STOP only +from `SCANNING` with a bound project. It emits only a non-executable digest and +wire metadata, uses QoS 2 with retain false, and declares automatic retry +forbidden. It has no MQTT publish dependency and cannot send a command. + +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 diff --git a/docs/adr/0012-device-bound-k1-command-authority.md b/docs/adr/0012-device-bound-k1-command-authority.md new file mode 100644 index 0000000..92b8478 --- /dev/null +++ b/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. diff --git a/plugins/xgrids-k1/frontend/src/api.ts b/plugins/xgrids-k1/frontend/src/api.ts index 9e103f5..e81b3af 100644 --- a/plugins/xgrids-k1/frontend/src/api.ts +++ b/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; diff --git a/src/k1link/device_plugins/xgrids_k1/facade.py b/src/k1link/device_plugins/xgrids_k1/facade.py index 67b71db..4213a7b 100644 --- a/src/k1link/device_plugins/xgrids_k1/facade.py +++ b/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, diff --git a/src/k1link/device_plugins/xgrids_k1/protocol/modeling_safety.py b/src/k1link/device_plugins/xgrids_k1/protocol/modeling_safety.py new file mode 100644 index 0000000..5e5e22d --- /dev/null +++ b/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") diff --git a/tests/test_xgrids_modeling_safety.py b/tests/test_xgrids_modeling_safety.py new file mode 100644 index 0000000..dca8f7e --- /dev/null +++ b/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] + )