From 36f0c93d2a92855893cd0efb5961e35670b1d12d Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Sat, 18 Jul 2026 17:19:21 +0300 Subject: [PATCH] fix(k1): gate control dialogue on live state --- README.md | 28 +- docs/01_IMPLEMENTATION_PLAN.md | 30 +- docs/05_K1_MQTT_STREAM_PROFILE.md | 60 +- .../0012-device-bound-k1-command-authority.md | 39 +- ...K1_CONTROL_ACCEPTANCE_20260718.redacted.md | 91 ++++ .../protocol/application_acceptance.py | 515 ++++++++++++++++-- .../protocol/application_bootstrap.py | 85 ++- .../xgrids_k1/protocol/application_mqtt.py | 228 +++++++- .../xgrids_k1/protocol/application_publish.py | 17 + .../xgrids_k1/protocol/modeling_control.py | 61 +++ tests/test_xgrids_application_acceptance.py | 191 ++++++- tests/test_xgrids_application_bootstrap.py | 16 + tests/test_xgrids_application_mqtt.py | 178 +++++- .../test_xgrids_modeling_control_protocol.py | 18 + 14 files changed, 1435 insertions(+), 122 deletions(-) create mode 100644 docs/lab/003_K1_CONTROL_ACCEPTANCE_20260718.redacted.md diff --git a/README.md b/README.md index 588a068..808218c 100644 --- a/README.md +++ b/README.md @@ -155,13 +155,26 @@ it together with the orchestrator. A separate, uninstalled physical-acceptance transport now reproduces MQTT 3.1.1 `clean_session=false`, keepalive 60, exact response subscriptions, QoS2 completion and the five response-gated batches. It consumes operation keys before publish and poisons unknown outcomes without -retry. An interactive Keychain administration command provisions through -Apple's hidden prompt without putting the authority in argv. Neither component -is wired to facade/UI; `vendor_writes_enabled` remains false. The first -operator-present physical attempt on 2026-07-18 emitted only the first six -bootstrap requests, then failed closed while correlating the third response -batch. The K1 remained `READY`; the `ModelingRequest` START was never emitted. -The temporary laboratory Keychain item was deleted after the attempt. +retry. Neither component is wired to facade/UI; `vendor_writes_enabled` remains +false. The first operator-present physical attempt on 2026-07-18 emitted only +the first six bootstrap requests, then failed closed while correlating the third +response batch; START was not emitted. A second one-shot attempt correlated all +ten pre-START requests and the START acknowledgement, but incorrectly collapsed +the whole exchange into 365 ms and ended the control lifecycle before the +retained post-START reads. K1 stayed in `SCAN_STARTING` for about 24 seconds, +reported `ALGORITHM_ERROR` twice, emitted no point/pose data and showed steady +red until an operator power cycle restored steady green. The collapsed executor +is now disabled. The offline acceptance executor maps connection requests 1–6, +workspace-entry request 7, project-prompt reads 8–10, user-confirmed START 11, +immediate status read 12 and state-gated reads 13–14 onto one continuously owned +MQTT session. No captured operator delay is a protocol timer: the final reads +wait for live `SCANNING`, project binding and `init_ready`, while STOP/save waits +for live `READY` plus physical standby confirmation. Standalone START and STOP +are both rejected. The same socket is serviced during every operator/UI wait, +and the short START/STOP permits are created only at their respective button +confirmations rather than at connection time. This executor remains uninstalled and is not wired to the +facade/UI; physical command acceptance is still paused. The temporary laboratory +Keychain item was deleted. This locked bootstrap is repeatable in the current workspace, not yet a standalone release install. The frontend consumes sibling `file:` packages from @@ -287,6 +300,7 @@ present. - [Live console and embedded Rerun runbook](docs/06_K1_LIVE_VIEWER.md) - [Mission Core monorepo and plugin boundary](docs/07_MISSION_CORE_MONOREPO.md) - [Owner-controlled LixelGO/iPhone observation](docs/08_LIXELGO_IPHONE_OBSERVATION.md) +- [K1 application-control physical acceptance](docs/lab/003_K1_CONTROL_ACCEPTANCE_20260718.redacted.md) - [Observation sessions, playback and workspace layout](docs/09_OBSERVATION_SESSIONS.md) - [Monorepo architecture decision](docs/adr/0002-mission-core-monorepo.md) - [Device plugin UI and runtime boundary](docs/adr/0003-device-plugin-ui-and-runtime-boundary.md) diff --git a/docs/01_IMPLEMENTATION_PLAN.md b/docs/01_IMPLEMENTATION_PLAN.md index b6a9b24..b2e677f 100644 --- a/docs/01_IMPLEMENTATION_PLAN.md +++ b/docs/01_IMPLEMENTATION_PLAN.md @@ -21,7 +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 (first physical bootstrap failed closed before START) — plugin v0.4.0 runtime remains dormant/write-disabled; the uninstalled acceptance transport completed one connection, six QoS2 bootstrap publishes and five correlated responses, then rejected one live batch-3 response; K1 remained READY, START was not emitted, retry is forbidden and the temporary Keychain item was deleted | +| K1 application control | PAUSE (accepted START followed by `ALGORITHM_ERROR`) — first live attempt stopped before START on a batch-3 correlation failure; a second attempt correlated all 10 pre-START requests plus START but collapsed UI stages into 365 ms and omitted the retained post-START/control-session lifecycle. The collapsed path and standalone START/STOP are disabled. An uninstalled offline executor now uses explicit UI checkpoints plus live DeviceStatus gates; no further physical write is authorized yet | | 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 @@ -61,7 +61,26 @@ required responses through batch 3, but the orchestrator raised `ApplicationBootstrapError` while correlating that batch. No START was emitted; live DeviceStatus stayed READY and the local point/pose counts remained zero. The local capture was sealed and the temporary Keychain item was deleted. -Physical START/STOP therefore remain untested; the button is the safe fallback. +The second physical attempt then correlated all ten bootstrap requests and one +START success response. It nevertheless did not reproduce the full LixelGO +lifecycle: all eleven publishes completed in 365 ms, the process did not retain +the control session through initialization, and it omitted the immediate and +post-initialization status reads present in both retained LixelGO START captures. +K1 reported `SCAN_STARTING` for about 24 seconds, emitted system error +`0x32040133` twice (the recovered `ALGORITHM_ERROR` namespace/value), returned +to `OTHER_STATUS`, produced zero point/pose frames and showed steady red. The +operator restored steady green through a normal power cycle. The legacy +`run_bootstrap()` path now rejects before emission. The offline acceptance path +maps requests 1–6 to connection, request 7 to explicit scan-workspace entry, +requests 8–10 to explicit project-prompt opening and request 11 to the operator's +START confirmation. Captured human delays are not replayed or treated as minimum +timeouts. After the immediate operation-12 read, operations 13–14 wait for the +same bound K1 to report `SCANNING`, a bound project and `init_ready=true`. +The same transport is then serviced until an explicit STOP request; post-STOP +ownership waits for live `READY` plus physical standby confirmation. System-error +and status reports are decoded into redacted safety state. This path is not +wired into the plugin runtime/UI and has only synthetic regression coverage; +physical STOP/save remains untested and the button is the 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; @@ -280,8 +299,9 @@ useful stream is decoded or structurally identified. - firmware or internal-Linux analysis; - physical end-to-end shared-timeline playback for a newly archived session containing point cloud plus one selected K1 camera; -- OS-independent plugin/edge authority provider, redacted live-response - diagnostics for bootstrap batch 3, a newly permitted operator-present full - START/STOP attempt and stable-artifact proof after STOP; +- OS-independent plugin/edge authority provider, PCAP-to-executor transcript + regression, runtime ownership integration, a separately reviewed and newly + permitted operator-present START/STOP attempt, and stable-artifact proof after + STOP; - long-running large-session WebViewer/WASM memory telemetry; - production retention, replication, encryption and cross-platform packaging. diff --git a/docs/05_K1_MQTT_STREAM_PROFILE.md b/docs/05_K1_MQTT_STREAM_PROFILE.md index f54cf97..4dc4b1c 100644 --- a/docs/05_K1_MQTT_STREAM_PROFILE.md +++ b/docs/05_K1_MQTT_STREAM_PROFILE.md @@ -188,14 +188,31 @@ repository shadow bootstrap reproduces all ten retained payloads and their topic order byte-for-byte. Time sync is the only mutation before START; the other nine requests are reads. -The retained request/response timing further divides the dialogue into five -bounded batches. DeviceInfo response establishes identity before later bound -requests. Required response barriers are correlated by topic, exact session, -vendor identity, application authority and numeric success. The initial -ModelingStatus request is intentionally not treated as a required synchronous -response: the clean cycle continued without one, and readiness belongs to the -live DeviceStatus gate. The offline orchestrator completes over the retained -payloads without substitution or skipped requests. +The first ten requests use five response-barrier batches, but those barriers are +not the whole application lifecycle. Two independent owner-operated START +captures contain the same 14-operation order: the ten requests above, START, +one bound `ModelingStatus` read, then a bound `DeviceInfo` plus +`ModelingStatus` refresh about 24–25 seconds after START. The first capture +reached START 202.423 seconds after its initial DeviceInfo; the faster capture +reached it after 54.130 seconds. This variation is operator/UI time, while the +stage order is invariant. In both captures the original control MQTT connection +remained open throughout initialization and the later refresh; no DISCONNECT or +UNSUBSCRIBE occurred. DeviceInfo response establishes identity before later +bound requests. Required response barriers are correlated by topic, exact +session, vendor identity, application authority and numeric success. The +initial and immediate post-START ModelingStatus reads are intentionally not +treated as mandatory synchronous responses. Readiness belongs to live +DeviceStatus plus point/pose observation. + +The UI/event correlation is also fixed by the retained capture. Requests 1–6 +follow successful data connection. Entering the scan workspace causes request +7. Opening the record/project-name prompt causes reads 8–10. The project name is +not sent by a separate save mutation: it first appears in the START payload when +the operator confirms the prompt. K1 then reports `SCAN_STARTING`; completion is +the live transition to `SCANNING` with a bound project and `init_ready=true`. +Operations 13–14 coincide with that state in both captures. The observed +24–25-second initialization duration is therefore evidence/telemetry, not a +hard-coded transition delay. The repository now contains an inert bounded encoder/response parser and a fail-closed device-status state machine for this exact profile. A live-only @@ -230,7 +247,32 @@ connection and emitted bootstrap ordinals 1–6. All six QoS 2 publishes complet and five required responses were collected, but exact correlation rejected one response in batch 3. The executor stopped before building or publishing START; K1 remained `READY`, project binding stayed absent and point/pose remained zero. -No automatic retry was attempted. +No automatic retry was attempted. A second explicitly permitted attempt then +correlated all ten bootstrap responses and one START success response, with 11 +QoS2 completions, but it compressed the exchange into 365 ms and ended without +the retained post-START reads or durable control-session ownership. K1 remained +in `SCAN_STARTING` for about 24 seconds, emitted two identical +`report/system_error` values `0x32040133`, returned to `OTHER_STATUS`, produced +no point/pose frames and showed steady red. The bounded system-error decoder maps +the `0x32040000 + 307` value to the recovered `ALGORITHM_ERROR` state. A normal +operator power cycle restored steady green. + +This disproves the earlier assumption that byte-matched pre-START payloads plus +a correlated START acknowledgement are sufficient acceptance. The legacy +collapsed `run_bootstrap()` executor now rejects before emission. Its +replacement is an uninstalled acceptance-only session owner. It requires the +observed connection, workspace-entry and project-prompt checkpoints instead of +wall-clock floors, sends the immediate post-START ModelingStatus read, services +the same MQTT socket until live `SCANNING + project + init_ready`, then emits the +DeviceInfo/ModelingStatus refresh. It continues servicing that socket until an +explicit STOP request and afterwards until live `READY` plus operator-confirmed +standby. System-error and DeviceStatus reports are decoded into redacted safety +state; a fault forbids further automatic action. Standalone START/STOP are +disabled. MQTT keepalive/report processing also continues while the operator is +between UI actions; the 15–120-second command permit is created at the actual +START or STOP confirmation, not at connection time. This path is +synthetic-test-only and remains outside facade/UI, so it +does not authorize a new physical attempt. The one-time Keychain administration command uses `security -w` as the final argument so Apple's own TTY prompt receives the authority. Mission Core never diff --git a/docs/adr/0012-device-bound-k1-command-authority.md b/docs/adr/0012-device-bound-k1-command-authority.md index 3df877e..2fdf285 100644 --- a/docs/adr/0012-device-bound-k1-command-authority.md +++ b/docs/adr/0012-device-bound-k1-command-authority.md @@ -37,10 +37,14 @@ K1 application control is exact-profile, live-bound and fail-closed. 3. The BLE-selected transport is followed by an unbound DeviceInfo exchange. Its live vendor ID, serial, activation and FW 3.0.2 facts form the transient device binding; a saved profile is optional metadata, not protocol authority. -4. The ten retained pre-START requests keep their exact order and response - boundaries. The only mutation is the observed time/timezone sync; the other - nine requests are reads. No request is batched, skipped or automatically - retried. +4. The complete retained START lifecycle keeps its 14-operation order, stage + boundaries and one continuous control MQTT session. The first ten operations + are pre-START; operation 11 is START; operation 12 is a bound ModelingStatus + read; operations 13–14 are the post-initialization DeviceInfo/ModelingStatus + refresh. The only preparatory mutation is time/timezone sync. No stage is + collapsed, skipped or automatically retried. Captured wall-clock gaps are + operator/UI timing, not protocol delays. Stage progression uses explicit UI + actions and live K1 state reports. 5. Live DeviceInfo binding must agree with the live DeviceStatus stream before START/STOP planning. Identity drift, malformed data, inactive equipment, profile mismatch or an unexpected lifecycle state fails closed. @@ -55,7 +59,7 @@ K1 application control is exact-profile, live-bound and fail-closed. quiescence/evidence sealing, READY plus steady green, then independent native project verification. -## Recovered pre-START sequence +## Recovered START sequence The clean cycle contains: @@ -70,6 +74,9 @@ The clean cycle contains: 9. bound `GetRtkAdvanceRequest`; 10. bound `DeviceInfoRequest`; 11. bound `ModelingRequest` START. +12. bound `ModelingStatusRequest`; +13. bound `DeviceInfoRequest` after static initialization; +14. bound `ModelingStatusRequest` after static initialization. Normal sessions are `${device_id_or_empty}:${MessageType}`. Time sync uses the captured special relation @@ -115,9 +122,20 @@ An unknown post-publish outcome can only be reconciled physically/status-wise; the same operation key can never be attempted again by that transport. The acceptance executor requires a 15–120 second single-action permit backed by -five explicit operator confirmations. START cannot skip the five recovered -bootstrap batches; STOP uses a separate permit. These types are not imported by -the facade, declared as plugin actions or represented by UI controls. +five explicit operator confirmations. After the failed 2026-07-18 physical +attempt proved that five response batches alone are insufficient, the collapsed +bootstrap entry point and standalone START/STOP were disabled. The uninstalled +replacement binds requests 1–6 to connection, request 7 to scan-workspace entry, +requests 8–10 to project-prompt entry and START to a separate operator permit. +There is no separate project-name mutation; the validated name is carried by +START. Operations 13–14 are gated by bound `SCANNING`, project presence and +`init_ready`, not by the observed 24–25-second duration. The same owner services +MQTT through explicit STOP and waits for live READY plus operator-confirmed +standby. Fault/status observations are redacted and any fault forbids automatic +continuation. The socket remains serviced during operator/UI waits, and each +short action permit is created at its button confirmation rather than at +connection time. These types are synthetic-test-only, not imported by the facade, +declared as plugin actions or represented by UI controls. `MacOSKeychainApplicationAuthorityProvisioner` invokes Apple's interactive `security add-generic-password ... -w` prompt with `-w` last. The private value @@ -133,10 +151,11 @@ transport: 1. provision the reviewed private authority into the fixed local Keychain item; 2. confirm battery/storage and select one K1 over BLE; -3. run the exact response-gated bootstrap and attest activated FW 3.0.2; +3. run the exact staged response-gated preparation and attest activated FW 3.0.2; 4. confirm READY identity against DeviceInfo and review shadow metadata; 5. send one START without retry and correlate its response; -6. observe calibration and first point/pose/camera data; +6. preserve the same control MQTT connection, issue the retained post-START + reads and observe calibration plus first point/pose/camera data; 7. send one STOP without retry and correlate its response; 8. seal local evidence while accepting the bounded stream tail; 9. wait for READY and steady green; diff --git a/docs/lab/003_K1_CONTROL_ACCEPTANCE_20260718.redacted.md b/docs/lab/003_K1_CONTROL_ACCEPTANCE_20260718.redacted.md new file mode 100644 index 0000000..c30cf31 --- /dev/null +++ b/docs/lab/003_K1_CONTROL_ACCEPTANCE_20260718.redacted.md @@ -0,0 +1,91 @@ +# Lab 003 — K1 application-control physical acceptance + +- Date: 2026-07-18 +- Device scope: one owner-controlled LixelKity K1, firmware `3.0.2` +- Result: **PAUSE — START acknowledgement is not lifecycle acceptance** + +This report contains no device identity, application authority, address, +credential, raw MQTT payload or project contents. Private evidence remains in +ignored session storage. + +## Safety boundary + +The operator was present and confirmed ownership, steady-green standby, +battery/storage, exact direct-LAN target and closed LixelGO. Every request used +QoS 2, retain false, one connection attempt and no automatic retry. The local +point/pose receiver was read-only and independent of the control client. + +## Attempt 1 + +The control client emitted bootstrap ordinals 1–6. All six QoS2 transactions +completed and five required responses arrived, but the batch-3 response did not +pass exact correlation. The executor stopped before START. K1 remained READY +and steady green; point/pose counts stayed zero. + +## Attempt 2 + +After adding redacted live-response diagnostics, one new explicit permit was +used. All ten pre-START requests and START completed QoS2; all required +responses, including the START response, passed identity/session/action/result +correlation. The eleven publishes completed in approximately `365 ms`. + +The device then reported `SCAN_STARTING` for approximately 24 seconds. No point +cloud or pose message arrived. It emitted two identical +`lixel/application/report/system_error` messages with code `0x32040133`, reset +to `OTHER_STATUS` without a project binding and showed a steady-red indicator. +The bounded decoder maps the observed namespace/value as +`0x32040000 + 307 = ALGORITHM_ERROR`. A normal operator power cycle restored +steady green. No guessed STOP or repeated START was sent. + +## Canonical-diff finding + +Re-audit of two independent owner-operated LixelGO START captures found the +same 14-operation order in both: + +1. unbound DeviceInfo; +2. unbound ModelingStatus; +3. unbound GetRtkAdvance; +4. bound time/timezone DeviceConfig; +5. bound DeviceInfo; +6. bound GetRtkAdvance; +7. bound GetNtripProfile; +8. bound GetCloudServerConfig; +9. bound GetRtkAdvance; +10. bound DeviceInfo; +11. bound START; +12. bound ModelingStatus; +13. bound DeviceInfo after initialization; +14. bound ModelingStatus after initialization. + +The first retained capture reached START at `+202.423 s`; the faster one at +`+54.130 s`. This is variable operator/UI time, not a fixed scanner timeout. +Both preserved the stage order, sent operation 12 within about one second of +START, sent operations 13–14 about 24–25 seconds after START and kept the same +control MQTT connection open throughout. Neither capture disconnected or +unsubscribed during initialization. + +The same re-audit shows that those wall-clock gaps are not protocol timers. +Requests 1–6 follow connection, request 7 follows explicit workspace entry, +requests 8–10 follow opening the project prompt, and START follows the operator's +confirmation. No separate project-name save request exists; the name first +appears in START. In both captures operations 13–14 coincide with K1's own +`SCANNING`, bound-project and `init_ready=true` DeviceStatus transition. + +Attempt 2 matched the first eleven payload/topic operations but violated the +larger lifecycle: it collapsed the UI stages, omitted operations 12–14 and did +not preserve a long-lived control-session owner. The evidence does not prove +which individual deviation caused the device algorithm failure, so none is +treated as the sole cause. + +## Code decision + +The legacy collapsed `run_bootstrap()` executor now rejects before emission. +The replacement acceptance-only executor uses explicit connection/workspace/ +project-prompt checkpoints with no captured-delay replay. It owns operations +1–14, gates the final refresh on live K1 state, surfaces redacted system errors, +keeps the same socket serviced through explicit STOP and waits for READY plus +operator-confirmed standby. The socket is also serviced between every UI action; +START/STOP permits begin only at their respective confirmations. Standalone +START/STOP are rejected. This remains an +uninstalled synthetic-test-only boundary; no new physical attempt is authorized +by this implementation. Physical double-click remains the canonical fallback. diff --git a/src/k1link/device_plugins/xgrids_k1/protocol/application_acceptance.py b/src/k1link/device_plugins/xgrids_k1/protocol/application_acceptance.py index c874a47..1d231b9 100644 --- a/src/k1link/device_plugins/xgrids_k1/protocol/application_acceptance.py +++ b/src/k1link/device_plugins/xgrids_k1/protocol/application_acceptance.py @@ -5,13 +5,18 @@ import math import threading import time from collections.abc import Callable, Collection, Sequence -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Literal, Protocol from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import ( + MODELING_STATUS_RESPONSE_TOPIC, ApplicationBootstrapError, + ApplicationControlAuthority, + CanonicalPostStartObservation, LiveDeviceControlBinding, ShadowApplicationBootstrapOrchestrator, + correlate_application_response, + decode_and_bind_device_info_response, ) from k1link.device_plugins.xgrids_k1.protocol.application_mqtt import ( MODELING_RESPONSE_TOPIC, @@ -32,6 +37,9 @@ from k1link.device_plugins.xgrids_k1.protocol.modeling_safety import ( MIN_ACCEPTANCE_PERMIT_SECONDS = 15.0 MAX_ACCEPTANCE_PERMIT_SECONDS = 120.0 +# Socket-pump quantum only. It is never used to advance a K1 dialogue stage. +CONTROL_NETWORK_PUMP_QUANTUM_SECONDS = 1.0 + class ApplicationAcceptanceError(RuntimeError): """The operator-present physical acceptance contract failed closed.""" @@ -45,6 +53,21 @@ class ApplicationBatchExchange(Protocol): required_response_topics: Collection[str], ) -> dict[str, bytes]: ... + def maintain_open_for( + self, + duration_seconds: float, + *, + allowed_response_topics: Collection[str] = (), + ) -> None: ... + + def scan_initialization_complete(self, binding: LiveDeviceControlBinding) -> bool: ... + + def pre_start_ready(self, binding: LiveDeviceControlBinding) -> bool: ... + + def standby_complete(self, binding: LiveDeviceControlBinding) -> bool: ... + + def validate_bound_status(self, binding: LiveDeviceControlBinding) -> None: ... + @dataclass(frozen=True, slots=True) class PhysicalAcceptanceChecklist: @@ -71,6 +94,27 @@ class PhysicalAcceptanceChecklist: raise ApplicationAcceptanceError("acceptance action must be an explicit ModelingAction") +@dataclass(frozen=True, slots=True) +class OperatorDialogueCheckpoint: + """One explicit LixelGO-equivalent UI transition; never a wall-clock gate.""" + + event: Literal["workspace-entered", "project-prompt-opened", "start-confirmed"] + operator_initiated: Literal[True] + owner_token: object = field(repr=False, compare=False) + + def __post_init__(self) -> None: + if self.event not in { + "workspace-entered", + "project-prompt-opened", + "start-confirmed", + }: + raise ApplicationAcceptanceError("operator dialogue checkpoint is unknown") + if self.operator_initiated is not True: + raise ApplicationAcceptanceError( + "operator dialogue checkpoint must be explicitly initiated" + ) + + class PhysicalAcceptancePermit: """Short, single-action capability that is consumed before MQTT publish.""" @@ -122,17 +166,37 @@ class PhysicalAcceptancePermit: class PhysicalAcceptanceDialogueExecutor: - """Drive the recovered barriers and exactly one permitted START or STOP.""" + """Drive only explicitly staged portions of the recovered K1 dialogue. + + The legacy implementation ran all ten pre-START reads/mutations in one + tight loop. That preserved topic order but did not preserve the observed + LixelGO lifecycle. The staged methods below intentionally expose the + connection, scan-workspace and project-prompt UI boundaries. They do not + replay wall-clock gaps from the operator capture. The canonical START/STOP + methods keep the same transport owner from operation 1 through the + operator-confirmed post-STOP standby boundary. The generic one-command + entry point remains disabled so START and STOP cannot be emitted by two + unrelated one-shot processes. + """ def __init__( self, transport: ApplicationBatchExchange, - permit: PhysicalAcceptancePermit, ) -> None: self._transport = transport - self._permit = permit self._bootstrap_complete = False self._command_complete = False + self._dialogue_stage = "new" + self._start_complete = False + self._stop_attempted = False + self._stop_complete = False + self._active_authority: ApplicationControlAuthority | None = None + self._active_binding: LiveDeviceControlBinding | None = None + self._prepared_binding: LiveDeviceControlBinding | None = None + self._checkpoint_owner = object() + self._issued_checkpoint: str | None = None + self._start_permit_snapshot: dict[str, object] | None = None + self._stop_permit_snapshot: dict[str, object] | None = None self._response_evidence: list[dict[str, object]] = [] self._correlation_failure: dict[str, object] | None = None @@ -140,60 +204,138 @@ class PhysicalAcceptanceDialogueExecutor: 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") + del orchestrator + raise ApplicationAcceptanceError( + "collapsed bootstrap is disabled; use the canonical staged dialogue" + ) - 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: - payload = responses[request.response_topic] - self._record_response_evidence( - phase="bootstrap", - operation_key=( - f"bootstrap:{request.ordinal}:{request.message_type}" - ), - response_topic=request.response_topic, - payload=payload, - ) - try: - orchestrator.accept_response(request.response_topic, payload) - except ApplicationBootstrapError as exc: - self._record_correlation_failure( - phase="bootstrap", - operation_key=( - f"bootstrap:{request.ordinal}:{request.message_type}" - ), - response_topic=request.response_topic, - reason=str(exc), - ) - raise ApplicationAcceptanceError( - "bootstrap response correlation failed for " - f"ordinal {request.ordinal} ({request.message_type}): {exc}" - ) from exc + def run_connection_stage( + self, + orchestrator: ShadowApplicationBootstrapOrchestrator, + ) -> LiveDeviceControlBinding: + """Emit retained ordinals 1-6 at control-session establishment.""" + if self._dialogue_stage != "new" or self._bootstrap_complete or self._command_complete: + raise ApplicationAcceptanceError("connection stage is not admissible now") + for expected_batch in (1, 2, 3): + self._exchange_bootstrap_batch(orchestrator, expected_batch=expected_batch) binding = orchestrator.binding if binding is None: - raise ApplicationAcceptanceError("bootstrap completed without a live device binding") + raise ApplicationAcceptanceError("connection stage produced no live device binding") + self._prepared_binding = binding + self._dialogue_stage = "connection-ready" + return binding + + def wait_for_operator_checkpoint( + self, + event: Literal["workspace-entered", "project-prompt-opened", "start-confirmed"], + event_observed: Callable[[], bool], + ) -> OperatorDialogueCheckpoint: + """Service the original socket until one exact operator UI event occurs.""" + + expected = { + "connection-ready": "workspace-entered", + "workspace-ready": "project-prompt-opened", + "project-ready": "start-confirmed", + }.get(self._dialogue_stage) + if event != expected or self._issued_checkpoint is not None: + raise ApplicationAcceptanceError( + "operator checkpoint does not match the canonical dialogue stage" + ) + binding = self._prepared_binding + if binding is None: + raise ApplicationAcceptanceError("canonical preparation binding is unavailable") + while not ( + self._transport.pre_start_ready(binding) and event_observed() + ): + self._transport.maintain_open_for( + CONTROL_NETWORK_PUMP_QUANTUM_SECONDS, + allowed_response_topics={MODELING_STATUS_RESPONSE_TOPIC}, + ) + self._transport.validate_bound_status(binding) + self._issued_checkpoint = event + return OperatorDialogueCheckpoint( + event=event, + operator_initiated=True, + owner_token=self._checkpoint_owner, + ) + + def run_workspace_entry_stage( + self, + orchestrator: ShadowApplicationBootstrapOrchestrator, + checkpoint: OperatorDialogueCheckpoint, + ) -> LiveDeviceControlBinding: + """Emit ordinal 7 only for the observed scan-workspace entry action.""" + + if self._dialogue_stage != "connection-ready": + raise ApplicationAcceptanceError("workspace entry requires the connection stage") + self._consume_checkpoint(checkpoint, expected="workspace-entered") + self._exchange_bootstrap_batch(orchestrator, expected_batch=4) + binding = orchestrator.binding + if binding is None: + raise ApplicationAcceptanceError("workspace entry lost the live device binding") + self._dialogue_stage = "workspace-ready" + return binding + + def run_project_prompt_stage( + self, + orchestrator: ShadowApplicationBootstrapOrchestrator, + checkpoint: OperatorDialogueCheckpoint, + ) -> LiveDeviceControlBinding: + """Emit ordinals 8-10 when the operator opens the project-name prompt.""" + + if self._dialogue_stage != "workspace-ready": + raise ApplicationAcceptanceError("project prompt requires workspace entry") + self._consume_checkpoint(checkpoint, expected="project-prompt-opened") + self._exchange_bootstrap_batch(orchestrator, expected_batch=5) + if not orchestrator.snapshot().bootstrap_complete: + raise ApplicationAcceptanceError("project prompt did not complete the transcript") + binding = orchestrator.binding + if binding is None: + raise ApplicationAcceptanceError("project prompt lost the live device binding") self._bootstrap_complete = True + self._dialogue_stage = "project-ready" return binding def execute_modeling(self, command: ShadowModelingCommand) -> ModelingResponse: + del command + raise ApplicationAcceptanceError( + "standalone modeling commands are disabled; use one canonical START-to-STOP session" + ) + + def execute_canonical_start( + self, + command: ShadowModelingCommand, + post_start: CanonicalPostStartObservation, + *, + authority: ApplicationControlAuthority, + binding: LiveDeviceControlBinding, + permit: PhysicalAcceptancePermit, + checkpoint: OperatorDialogueCheckpoint, + ) -> ModelingResponse: + """Execute retained operations 11-14 on one continuously serviced socket.""" + + if command.action is not ModelingAction.START: + raise ApplicationAcceptanceError("canonical START executor requires START") 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) + if not self._bootstrap_complete or self._dialogue_stage != "project-ready": + raise ApplicationAcceptanceError("START requires the complete staged preparation") + if [request.ordinal for request in post_start.requests] != [12, 13, 14]: + raise ApplicationAcceptanceError("post-START transcript ordinals are invalid") + self._consume_checkpoint(checkpoint, expected="start-confirmed") + self._require_command_identity(command, authority=authority, binding=binding) + if not self._transport.pre_start_ready(binding): + raise ApplicationAcceptanceError( + "canonical START requires live READY with no bound project" + ) + if permit.action is not ModelingAction.START: + raise ApplicationAcceptanceError("canonical START requires a fresh START permit") + + permit.consume(ModelingAction.START) + self._start_permit_snapshot = permit.snapshot() self._command_complete = True + self._dialogue_stage = "start-attempted" responses = self._transport.exchange_batch_once( [OneShotPublishEnvelope.from_modeling_command(command)], required_response_topics={MODELING_RESPONSE_TOPIC}, @@ -201,29 +343,218 @@ class PhysicalAcceptanceDialogueExecutor: payload = responses[MODELING_RESPONSE_TOPIC] self._record_response_evidence( phase="modeling", - operation_key=f"modeling:{command.action.name.casefold()}", + operation_key="modeling:start", response_topic=MODELING_RESPONSE_TOPIC, payload=payload, ) try: - return correlate_modeling_response(payload, command.command) + response = correlate_modeling_response(payload, command.command) except ModelingProtocolError as exc: self._record_correlation_failure( phase="modeling", - operation_key=f"modeling:{command.action.name.casefold()}", + operation_key="modeling:start", response_topic=MODELING_RESPONSE_TOPIC, reason=str(exc), ) raise ApplicationAcceptanceError( - f"modeling response correlation failed for {command.action.name}: {exc}" + f"modeling response correlation failed for START: {exc}" ) from exc + immediate = post_start.immediate_modeling_status + self._transport.exchange_batch_once( + [OneShotPublishEnvelope.from_dialogue_request(immediate)], + required_response_topics=(), + ) + self._dialogue_stage = "initializing" + while not self._transport.scan_initialization_complete(binding): + self._transport.maintain_open_for( + CONTROL_NETWORK_PUMP_QUANTUM_SECONDS, + allowed_response_topics={MODELING_STATUS_RESPONSE_TOPIC}, + ) + + refresh = post_start.post_initialization_refresh + required_topics = {request.response_topic for request in refresh} + refresh_responses = self._transport.exchange_batch_once( + [OneShotPublishEnvelope.from_dialogue_request(request) for request in refresh], + required_response_topics=required_topics, + ) + for request in refresh: + refresh_payload = refresh_responses[request.response_topic] + operation_key = f"dialogue:{request.ordinal}:{request.message_type}" + self._record_response_evidence( + phase="post-start", + operation_key=operation_key, + response_topic=request.response_topic, + payload=refresh_payload, + ) + try: + if request.message_type == "DeviceInfoRequest": + observed = decode_and_bind_device_info_response( + refresh_payload, + authority, + expected_session_id=request.session_id, + expected_vendor_device_id=binding.vendor_device_id, + ) + if observed.binding != binding: + raise ApplicationBootstrapError( + "live DeviceInfo facts changed after START" + ) + else: + correlate_application_response( + refresh_payload, + request, + authority, + live_binding=binding, + ) + except ApplicationBootstrapError as exc: + self._record_correlation_failure( + phase="post-start", + operation_key=operation_key, + response_topic=request.response_topic, + reason=str(exc), + ) + raise ApplicationAcceptanceError( + f"post-START response correlation failed for {operation_key}: {exc}" + ) from exc + self._dialogue_stage = "post-initialization-observed" + self._start_complete = True + self._active_authority = authority + self._active_binding = binding + self._prepared_binding = None + return response + + def maintain_active_until_stop_requested( + self, + stop_requested: Callable[[], bool], + ) -> None: + """Continuously service the original socket until the operator requests STOP.""" + + if self._dialogue_stage != "post-initialization-observed" or not self._start_complete: + raise ApplicationAcceptanceError( + "active control ownership requires the complete post-START dialogue" + ) + binding = self._active_binding + if binding is None: + raise ApplicationAcceptanceError("canonical START binding is no longer available") + while not stop_requested(): + self._transport.maintain_open_for( + CONTROL_NETWORK_PUMP_QUANTUM_SECONDS, + allowed_response_topics={MODELING_STATUS_RESPONSE_TOPIC}, + ) + if not self._transport.scan_initialization_complete(binding): + raise ApplicationAcceptanceError( + "K1 left the bound SCANNING state before canonical STOP" + ) + self._dialogue_stage = "stop-requested" + + def execute_canonical_stop( + self, + command: ShadowModelingCommand, + permit: PhysicalAcceptancePermit, + ) -> ModelingResponse: + """Emit retained STOP on the same socket and with a separate permit.""" + + if command.action is not ModelingAction.STOP: + raise ApplicationAcceptanceError("canonical STOP executor requires STOP") + if self._dialogue_stage != "stop-requested" or not self._start_complete: + raise ApplicationAcceptanceError( + "STOP requires continuous ownership from the canonical START session" + ) + if self._stop_attempted: + raise ApplicationAcceptanceError("canonical STOP was already attempted") + authority = self._active_authority + binding = self._active_binding + if authority is None or binding is None: + raise ApplicationAcceptanceError("canonical START binding is no longer available") + self._require_command_identity(command, authority=authority, binding=binding) + if not self._transport.scan_initialization_complete(binding): + raise ApplicationAcceptanceError( + "canonical STOP requires the bound K1 to still report SCANNING" + ) + if permit.action is not ModelingAction.STOP: + raise ApplicationAcceptanceError("canonical STOP requires a separate STOP permit") + + permit.consume(ModelingAction.STOP) + self._stop_permit_snapshot = permit.snapshot() + self._stop_attempted = True + self._dialogue_stage = "stop-attempted" + responses = self._transport.exchange_batch_once( + [OneShotPublishEnvelope.from_modeling_command(command)], + required_response_topics={MODELING_RESPONSE_TOPIC}, + ) + payload = responses[MODELING_RESPONSE_TOPIC] + self._record_response_evidence( + phase="modeling", + operation_key="modeling:stop", + response_topic=MODELING_RESPONSE_TOPIC, + payload=payload, + ) + try: + response = correlate_modeling_response(payload, command.command) + except ModelingProtocolError as exc: + self._record_correlation_failure( + phase="modeling", + operation_key="modeling:stop", + response_topic=MODELING_RESPONSE_TOPIC, + reason=str(exc), + ) + raise ApplicationAcceptanceError( + f"modeling response correlation failed for STOP: {exc}" + ) from exc + self._stop_complete = True + self._dialogue_stage = "stop-acknowledged" + return response + + def maintain_post_stop_until_standby_confirmed( + self, + standby_confirmed: Callable[[], bool], + ) -> None: + """Keep servicing control reports through save and physical standby. + + The retained capture does not expose a protocol timer that proves save + completion. The session therefore remains owned until an explicit + operator/status gate confirms standby and never closes itself merely + because a captured wall-clock duration elapsed. + """ + + if self._dialogue_stage != "stop-acknowledged" or not self._stop_complete: + raise ApplicationAcceptanceError( + "post-STOP ownership requires a correlated canonical STOP" + ) + binding = self._active_binding + if binding is None: + raise ApplicationAcceptanceError("canonical START binding is no longer available") + while not ( + self._transport.standby_complete(binding) and standby_confirmed() + ): + self._transport.maintain_open_for( + CONTROL_NETWORK_PUMP_QUANTUM_SECONDS, + allowed_response_topics={MODELING_STATUS_RESPONSE_TOPIC}, + ) + self._dialogue_stage = "standby-confirmed" + self._active_authority = None + self._active_binding = None + 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(), + "start_attempted": self._command_complete, + "start_complete": self._start_complete, + "stop_attempted": self._stop_attempted, + "stop_complete": self._stop_complete, + "dialogue_stage": self._dialogue_stage, + "start_permit": ( + dict(self._start_permit_snapshot) + if self._start_permit_snapshot is not None + else None + ), + "stop_permit": ( + dict(self._stop_permit_snapshot) + if self._stop_permit_snapshot is not None + else None + ), "response_evidence": tuple(dict(item) for item in self._response_evidence), "correlation_failure": ( dict(self._correlation_failure) @@ -233,10 +564,86 @@ class PhysicalAcceptanceDialogueExecutor: "automatic_retry": False, } + @staticmethod + def _require_command_identity( + command: ShadowModelingCommand, + *, + authority: ApplicationControlAuthority, + binding: LiveDeviceControlBinding, + ) -> None: + header = command.command.header + if ( + header.device_id != binding.vendor_device_id + or header.openapi_key != authority.openapi_key + ): + raise ApplicationAcceptanceError( + "modeling command identity does not match the live canonical session" + ) + + def _consume_checkpoint( + self, + checkpoint: OperatorDialogueCheckpoint, + *, + expected: str, + ) -> None: + if ( + checkpoint.owner_token is not self._checkpoint_owner + or checkpoint.event != expected + or self._issued_checkpoint != expected + ): + raise ApplicationAcceptanceError( + "operator checkpoint was not issued by this canonical session" + ) + self._issued_checkpoint = None + + def _exchange_bootstrap_batch( + self, + orchestrator: ShadowApplicationBootstrapOrchestrator, + *, + expected_batch: int, + ) -> None: + snapshot = orchestrator.snapshot() + if snapshot.current_batch != expected_batch or snapshot.batch_issued: + raise ApplicationAcceptanceError( + f"canonical bootstrap expected batch {expected_batch}" + ) + 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 not request.response_required: + continue + payload = responses[request.response_topic] + operation_key = f"bootstrap:{request.ordinal}:{request.message_type}" + self._record_response_evidence( + phase="bootstrap", + operation_key=operation_key, + response_topic=request.response_topic, + payload=payload, + ) + try: + orchestrator.accept_response(request.response_topic, payload) + except ApplicationBootstrapError as exc: + self._record_correlation_failure( + phase="bootstrap", + operation_key=operation_key, + response_topic=request.response_topic, + reason=str(exc), + ) + raise ApplicationAcceptanceError( + "bootstrap response correlation failed for " + f"ordinal {request.ordinal} ({request.message_type}): {exc}" + ) from exc + def _record_response_evidence( self, *, - phase: Literal["bootstrap", "modeling"], + phase: Literal["bootstrap", "modeling", "post-start"], operation_key: str, response_topic: str, payload: bytes, @@ -254,7 +661,7 @@ class PhysicalAcceptanceDialogueExecutor: def _record_correlation_failure( self, *, - phase: Literal["bootstrap", "modeling"], + phase: Literal["bootstrap", "modeling", "post-start"], operation_key: str, response_topic: str, reason: str, diff --git a/src/k1link/device_plugins/xgrids_k1/protocol/application_bootstrap.py b/src/k1link/device_plugins/xgrids_k1/protocol/application_bootstrap.py index 23ec2db..5ff4b45 100644 --- a/src/k1link/device_plugins/xgrids_k1/protocol/application_bootstrap.py +++ b/src/k1link/device_plugins/xgrids_k1/protocol/application_bootstrap.py @@ -129,7 +129,7 @@ class ApplicationRequestHeader: @dataclass(frozen=True, slots=True) class EncodedApplicationRequest: ordinal: int - phase: Literal["identity-discovery", "bound-preparation"] + phase: Literal["identity-discovery", "bound-preparation", "post-start-observation"] message_type: str topic: str response_topic: str @@ -182,6 +182,21 @@ class ShadowApplicationBootstrap: } +@dataclass(frozen=True, slots=True) +class CanonicalPostStartObservation: + """Exact retained operations 12-14 following a successful START publish.""" + + immediate_modeling_status: EncodedApplicationRequest = field(repr=False) + post_initialization_refresh: tuple[ + EncodedApplicationRequest, + EncodedApplicationRequest, + ] = field(repr=False) + + @property + def requests(self) -> tuple[EncodedApplicationRequest, ...]: + return (self.immediate_modeling_status, *self.post_initialization_refresh) + + @dataclass(frozen=True, slots=True) class DeviceInfoResponse: binding: LiveDeviceControlBinding = field(repr=False) @@ -384,6 +399,74 @@ def build_shadow_bootstrap( return ShadowApplicationBootstrap(tuple(requests)) +def build_canonical_post_start_observation( + authority: ApplicationControlAuthority, + binding: LiveDeviceControlBinding, +) -> CanonicalPostStartObservation: + """Build retained operations 12-14 without granting publish authority.""" + + if not binding.ready_for_reviewed_profile: + raise ApplicationBootstrapError( + "live DeviceInfo does not attest the reviewed activated FW 3.0.2 profile" + ) + + def build( + ordinal: Literal[12, 13, 14], + message_type: Literal["DeviceInfoRequest", "ModelingStatusRequest"], + topic: str, + response_topic: str, + *, + response_required: bool, + ) -> EncodedApplicationRequest: + header = ApplicationRequestHeader( + message_type=message_type, + authority=authority, + binding=binding, + ) + payload = _bytes_field(1, _encode_header(header)) + _check_payload(payload, "post-START application request") + return EncodedApplicationRequest( + ordinal=ordinal, + phase="post-start-observation", + message_type=message_type, + topic=topic, + response_topic=response_topic, + payload=payload, + payload_sha256=hashlib.sha256(payload).hexdigest(), + payload_bytes=len(payload), + mutates_device=False, + requires_live_binding=True, + response_required=response_required, + session_id=header.session_id, + vendor_device_id=binding.vendor_device_id, + ) + + immediate = build( + 12, + "ModelingStatusRequest", + MODELING_STATUS_REQUEST_TOPIC, + MODELING_STATUS_RESPONSE_TOPIC, + response_required=False, + ) + refresh = ( + build( + 13, + "DeviceInfoRequest", + DEVICE_INFO_REQUEST_TOPIC, + DEVICE_INFO_RESPONSE_TOPIC, + response_required=True, + ), + build( + 14, + "ModelingStatusRequest", + MODELING_STATUS_REQUEST_TOPIC, + MODELING_STATUS_RESPONSE_TOPIC, + response_required=True, + ), + ) + return CanonicalPostStartObservation(immediate, refresh) + + def decode_and_bind_device_info_response( payload: bytes, authority: ApplicationControlAuthority, diff --git a/src/k1link/device_plugins/xgrids_k1/protocol/application_mqtt.py b/src/k1link/device_plugins/xgrids_k1/protocol/application_mqtt.py index 853787a..9362d76 100644 --- a/src/k1link/device_plugins/xgrids_k1/protocol/application_mqtt.py +++ b/src/k1link/device_plugins/xgrids_k1/protocol/application_mqtt.py @@ -22,11 +22,17 @@ from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import ( GET_NTRIP_PROFILE_RESPONSE_TOPIC, GET_RTK_ADVANCE_RESPONSE_TOPIC, MODELING_STATUS_RESPONSE_TOPIC, + LiveDeviceControlBinding, ) from k1link.device_plugins.xgrids_k1.protocol.application_publish import ( OneShotPublishEnvelope, ) +from k1link.device_plugins.xgrids_k1.protocol.modeling_control import ( + decode_device_status_report, + decode_system_error_report, +) from k1link.device_plugins.xgrids_k1.protocol.modeling_safety import ( + DEVICE_STATUS_TOPIC, MODELING_REQUEST_TOPIC, ) @@ -35,7 +41,9 @@ CONTROL_KEEPALIVE_SECONDS = 60 CONTROL_CONNECT_TIMEOUT_SECONDS = 10.0 CONTROL_EXCHANGE_TIMEOUT_SECONDS = 5.0 CONTROL_LOOP_INTERVAL_SECONDS = 0.05 +MAX_CONTROL_MAINTAIN_SECONDS = 30.0 MAX_CONTROL_RESPONSE_BYTES = 64 * 1024 +SYSTEM_ERROR_TOPIC = "lixel/application/report/system_error" # The retained LixelGO control connection issued these three SUBSCRIBE packets # in this exact order. Point-cloud subscriptions belonged to a separate client. @@ -47,8 +55,8 @@ CONTROL_SUBSCRIPTION_GROUPS: tuple[tuple[tuple[str, int], ...], ...] = ( ("PrePathArray", 0), ("ScanStatus", 0), ("lixel/application/report/lio_pose", 0), - ("lixel/application/report/device_status", 0), - ("lixel/application/report/system_error", 0), + (DEVICE_STATUS_TOPIC, 0), + (SYSTEM_ERROR_TOPIC, 0), (DEVICE_INFO_RESPONSE_TOPIC, 0), ), ( @@ -144,6 +152,10 @@ class ApplicationCommandOutcomeUnknown(RuntimeError): """A request may have reached the K1 and must never be retried automatically.""" +class ApplicationControlDeviceFault(RuntimeError): + """Live status made further automatic control inadmissible.""" + + @dataclass(frozen=True, slots=True) class ApplicationMqttTransportSnapshot: state: str @@ -153,6 +165,14 @@ class ApplicationMqttTransportSnapshot: qos2_completions: int correlated_responses: int ignored_known_responses: int + device_status_reports: int + system_error_reports: int + report_decode_errors: int + latest_device_session_state: str | None + latest_device_project_bound: bool | None + latest_device_init_ready: bool | None + latest_system_error_code: int | None + latest_system_error_state: str | None operation_keys_consumed: int clean_session: bool = False keepalive_seconds: int = CONTROL_KEEPALIVE_SECONDS @@ -169,6 +189,14 @@ class ApplicationMqttTransportSnapshot: "qos2_completions": self.qos2_completions, "correlated_responses": self.correlated_responses, "ignored_known_responses": self.ignored_known_responses, + "device_status_reports": self.device_status_reports, + "system_error_reports": self.system_error_reports, + "report_decode_errors": self.report_decode_errors, + "latest_device_session_state": self.latest_device_session_state, + "latest_device_project_bound": self.latest_device_project_bound, + "latest_device_init_ready": self.latest_device_init_ready, + "latest_system_error_code": self.latest_system_error_code, + "latest_system_error_state": self.latest_system_error_state, "operation_keys_consumed": self.operation_keys_consumed, "clean_session": self.clean_session, "keepalive_seconds": self.keepalive_seconds, @@ -230,6 +258,17 @@ class ReviewedApplicationMqttTransport: self._qos2_completions = 0 self._correlated_responses = 0 self._ignored_known_responses = 0 + self._device_status_reports = 0 + self._system_error_reports = 0 + self._report_decode_errors = 0 + self._latest_device_session_state: str | None = None + self._latest_device_project_bound: bool | None = None + self._latest_device_init_ready: bool | None = None + self._latest_device_id: str | None = None + self._latest_device_serial: str | None = None + self._latest_device_fault = False + self._latest_system_error_code: int | None = None + self._latest_system_error_state: str | None = None def open(self) -> ApplicationMqttTransportSnapshot: with self._lock: @@ -267,7 +306,12 @@ class ReviewedApplicationMqttTransport: 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: + if not required and any( + envelope.topic != "lixel/application/request/modeling_status" + for envelope in batch + ): + raise ValueError("response-free exchange is limited to retained ModelingStatus reads") + if 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): @@ -333,6 +377,43 @@ class ReviewedApplicationMqttTransport: self._correlated_responses += len(responses) return responses + def maintain_open_for( + self, + duration_seconds: float, + *, + allowed_response_topics: Collection[str] = (), + ) -> None: + """Continuously service the same control socket across static initialization.""" + + if ( + not isinstance(duration_seconds, (int, float)) + or isinstance(duration_seconds, bool) + or not math.isfinite(duration_seconds) + or not 0 < duration_seconds <= MAX_CONTROL_MAINTAIN_SECONDS + ): + raise ValueError("control maintain duration is outside the reviewed bound") + allowed = frozenset(allowed_response_topics) + if not allowed <= frozenset({MODELING_STATUS_RESPONSE_TOPIC}): + raise ValueError("control maintain response allowlist exceeds retained dialogue") + with self._lock: + if self._state != "ready" or not self._connected or not self._subscribed: + raise ApplicationMqttTransportError("control transport is not ready") + deadline = self._monotonic() + float(duration_seconds) + client = self._require_client() + while self._monotonic() < deadline: + self._discard_allowed_responses(allowed) + with self._lock: + callback_error = self._callback_error + if callback_error is not None: + self._fail_after_publish(callback_error) + try: + result = client.loop(timeout=CONTROL_LOOP_INTERVAL_SECONDS) + except (OSError, RuntimeError, ValueError) as exc: + self._fail_after_publish("control MQTT network loop failed", exc) + if result != mqtt.MQTT_ERR_SUCCESS: + self._fail_after_publish("control MQTT network loop returned an error") + self._discard_allowed_responses(allowed) + def close(self) -> None: with self._lock: if self._state == "closed": @@ -355,6 +436,43 @@ class ReviewedApplicationMqttTransport: if self._state not in {"poisoned", "failed"}: self._state = "closed" + def scan_initialization_complete(self, binding: LiveDeviceControlBinding) -> bool: + """Return true only for the bound SCANNING/project/init-ready state.""" + + with self._lock: + self._require_healthy_bound_status_locked(binding) + return ( + self._latest_device_session_state == "scanning" + and self._latest_device_project_bound is True + and self._latest_device_init_ready is True + ) + + def pre_start_ready(self, binding: LiveDeviceControlBinding) -> bool: + """Return true only when the bound K1 is READY with no current project.""" + + with self._lock: + self._require_healthy_bound_status_locked(binding) + return ( + self._latest_device_session_state == "ready" + and self._latest_device_project_bound is False + ) + + def standby_complete(self, binding: LiveDeviceControlBinding) -> bool: + """Return true only when the same K1 reports unbound READY after STOP.""" + + with self._lock: + self._require_healthy_bound_status_locked(binding) + return ( + self._latest_device_session_state == "ready" + and self._latest_device_project_bound is False + ) + + def validate_bound_status(self, binding: LiveDeviceControlBinding) -> None: + """Fail closed on report faults or identity drift without advancing state.""" + + with self._lock: + self._require_healthy_bound_status_locked(binding) + def snapshot(self) -> ApplicationMqttTransportSnapshot: with self._lock: return ApplicationMqttTransportSnapshot( @@ -365,6 +483,14 @@ class ReviewedApplicationMqttTransport: qos2_completions=self._qos2_completions, correlated_responses=self._correlated_responses, ignored_known_responses=self._ignored_known_responses, + device_status_reports=self._device_status_reports, + system_error_reports=self._system_error_reports, + report_decode_errors=self._report_decode_errors, + latest_device_session_state=self._latest_device_session_state, + latest_device_project_bound=self._latest_device_project_bound, + latest_device_init_ready=self._latest_device_init_ready, + latest_system_error_code=self._latest_system_error_code, + latest_system_error_state=self._latest_system_error_state, operation_keys_consumed=len(self._consumed_operation_keys), ) @@ -449,10 +575,12 @@ class ReviewedApplicationMqttTransport: 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: + if message.topic in CONTROL_REPORT_TOPICS: + self._observe_control_report(message.topic, payload) + with self._lock: self._ignored_known_responses += 1 - else: + else: + with self._lock: self._messages.append((message.topic, payload)) def on_disconnect( @@ -474,6 +602,77 @@ class ReviewedApplicationMqttTransport: client.on_message = on_message client.on_disconnect = on_disconnect + def _observe_control_report(self, topic: str, payload: bytes) -> None: + if topic == DEVICE_STATUS_TOPIC: + try: + device_report = decode_device_status_report(payload) + except ValueError: + with self._lock: + self._report_decode_errors += 1 + return + with self._lock: + self._device_status_reports += 1 + self._latest_device_session_state = ( + device_report.session_state.name.casefold() + if device_report.session_state is not None + else None + ) + self._latest_device_project_bound = bool(device_report.project_id) + self._latest_device_init_ready = device_report.init_ready + self._latest_device_id = ( + device_report.header.device_id + if device_report.header is not None + else None + ) + self._latest_device_serial = device_report.device_sn + self._latest_device_fault = bool( + device_report.session_state is not None + and device_report.session_state.is_fault + ) + return + if topic == SYSTEM_ERROR_TOPIC: + try: + system_error_report = decode_system_error_report(payload) + except ValueError: + with self._lock: + self._report_decode_errors += 1 + return + with self._lock: + self._system_error_reports += 1 + self._latest_system_error_code = system_error_report.error_code + self._latest_system_error_state = ( + system_error_report.session_state.name.casefold() + if system_error_report.session_state is not None + else None + ) + + def _require_healthy_bound_status_locked( + self, + binding: LiveDeviceControlBinding, + ) -> None: + if self._report_decode_errors: + raise ApplicationControlDeviceFault( + "K1 control report decoding failed; no further automatic action is admissible" + ) + if self._system_error_reports or self._latest_device_fault: + state = self._latest_system_error_state or self._latest_device_session_state + raise ApplicationControlDeviceFault( + f"K1 reported a control fault state: {state or 'unknown'}" + ) + if self._device_status_reports == 0: + return + if self._latest_device_id is None or self._latest_device_serial is None: + raise ApplicationControlDeviceFault( + "K1 control status omitted the live device identity" + ) + if ( + self._latest_device_id != binding.vendor_device_id + or self._latest_device_serial != binding.device_serial + ): + raise ApplicationControlDeviceFault( + "K1 control status identity drifted from the live DeviceInfo binding" + ) + def _subscribe_next_group(self, client: mqtt.Client) -> None: with self._lock: group_index = self._subscription_group_index @@ -552,6 +751,23 @@ class ReviewedApplicationMqttTransport: self.close() raise ApplicationCommandOutcomeUnknown(ambiguous_message) + def _discard_allowed_responses(self, allowed: frozenset[str]) -> None: + ambiguous_message: str | None = None + with self._lock: + while self._messages: + topic, _payload = self._messages.popleft() + if topic in allowed: + self._ignored_known_responses += 1 + continue + self._poison_locked("unexpected response during retained control hold") + ambiguous_message = ( + "unexpected control response during retained control-session hold" + ) + break + 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: diff --git a/src/k1link/device_plugins/xgrids_k1/protocol/application_publish.py b/src/k1link/device_plugins/xgrids_k1/protocol/application_publish.py index 567def0..f57c55c 100644 --- a/src/k1link/device_plugins/xgrids_k1/protocol/application_publish.py +++ b/src/k1link/device_plugins/xgrids_k1/protocol/application_publish.py @@ -69,6 +69,23 @@ class OneShotPublishEnvelope: retain=request.retain, ) + @classmethod + def from_dialogue_request( + cls, + request: EncodedApplicationRequest, + ) -> OneShotPublishEnvelope: + if request.phase != "post-start-observation" or request.ordinal not in {12, 13, 14}: + raise ValueError("application request is not a retained post-START operation") + return cls( + operation_key=f"dialogue:{request.ordinal}:{request.message_type}", + topic=request.topic, + payload=request.payload, + payload_sha256=request.payload_sha256, + payload_bytes=request.payload_bytes, + qos=request.qos, + retain=request.retain, + ) + @classmethod def from_modeling_command( cls, diff --git a/src/k1link/device_plugins/xgrids_k1/protocol/modeling_control.py b/src/k1link/device_plugins/xgrids_k1/protocol/modeling_control.py index abdd428..9d848fb 100644 --- a/src/k1link/device_plugins/xgrids_k1/protocol/modeling_control.py +++ b/src/k1link/device_plugins/xgrids_k1/protocol/modeling_control.py @@ -21,6 +21,7 @@ MODELING_SESSION_SUFFIX = ":ModelingRequest" OPENAPI_RESULT_BASE = 302_252_032 OPENAPI_SUCCESS = OPENAPI_RESULT_BASE + 1 MODELING_STATE_BASE = 302_252_032 +SYSTEM_ERROR_STATE_BASE = 0x3204_0000 class ModelingProtocolError(ValueError): @@ -202,6 +203,15 @@ class DeviceStatusReport: rtk_status_present: bool +@dataclass(frozen=True, slots=True) +class SystemErrorReport: + """Bounded K1 system-error report with identity kept out of repr.""" + + header: ObservedHeader | None = field(repr=False) + error_code: int + session_state: SessionState | None + + def encode_modeling_start( header: CommandHeaderIdentity, *, @@ -391,6 +401,57 @@ def session_state_from_code(code: int) -> SessionState | None: return None +def decode_system_error_report( + payload: bytes, *, max_payload_bytes: int = MAX_CONTROL_PAYLOAD_BYTES +) -> SystemErrorReport: + """Decode the observed ``report/system_error`` envelope without guessing text.""" + + _check_payload_bound(payload, max_payload_bytes, "SystemErrorReport") + header: ObservedHeader | None = None + error_code: int | None = None + seen: set[int] = set() + try: + for proto_field in iter_fields(payload, max_fields=64): + if proto_field.number == 1: + _mark_once(seen, 1, "system_error.header") + header = _decode_observed_header( + _bytes_value(proto_field, "system_error.header"), require_identity=False + ) + elif proto_field.number == 15: + _mark_once(seen, 15, "system_error.error") + nested_seen: set[int] = set() + for nested in iter_fields( + _bytes_value(proto_field, "system_error.error"), max_fields=8 + ): + if nested.number == 1: + _mark_once(nested_seen, 1, "system_error.error.code") + error_code = _uint_value( + nested, + "system_error.error.code", + max_value=0xFFFF_FFFF, + ) + except ProtobufWireError as exc: + raise ModelingProtocolError(f"invalid SystemErrorReport: {exc}") from exc + if error_code is None: + raise ModelingProtocolError("SystemErrorReport has no error code") + return SystemErrorReport( + header=header, + error_code=error_code, + session_state=system_error_state_from_code(error_code), + ) + + +def system_error_state_from_code(code: int) -> SessionState | None: + """Map the observed system-error namespace while preserving unknown codes.""" + + if code < SYSTEM_ERROR_STATE_BASE: + return None + try: + return SessionState(code - SYSTEM_ERROR_STATE_BASE) + except ValueError: + return None + + def _encode_command_header(header: CommandHeaderIdentity) -> bytes: # seq/stamp/scaler were default-valued and omitted in the reviewed command # requests. The three identities must be supplied explicitly by a caller. diff --git a/tests/test_xgrids_application_acceptance.py b/tests/test_xgrids_application_acceptance.py index 96ee05b..c1eef78 100644 --- a/tests/test_xgrids_application_acceptance.py +++ b/tests/test_xgrids_application_acceptance.py @@ -7,6 +7,7 @@ import pytest from k1link.device_plugins.xgrids_k1.protocol.application_acceptance import ( ApplicationAcceptanceError, + OperatorDialogueCheckpoint, PhysicalAcceptanceChecklist, PhysicalAcceptanceDialogueExecutor, PhysicalAcceptancePermit, @@ -15,6 +16,7 @@ from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import ( DEVICE_CONFIG_RESPONSE_TOPIC, ApplicationControlAuthority, ShadowApplicationBootstrapOrchestrator, + build_canonical_post_start_observation, ) from k1link.device_plugins.xgrids_k1.protocol.application_mqtt import ( MODELING_RESPONSE_TOPIC, @@ -91,8 +93,15 @@ def _modeling_response(action: ModelingAction) -> bytes: class SyntheticAcceptanceTransport: - def __init__(self) -> None: + def __init__(self, clock: FakeClock | None = None) -> None: self.batches: list[tuple[str, ...]] = [] + self.maintain_calls: list[tuple[float, tuple[str, ...]]] = [] + self.clock = clock + self.pre_start_maintain_calls = 0 + self.active_maintain_calls = 0 + self.post_stop_maintain_calls = 0 + self.start_emitted = False + self.stop_emitted = False def exchange_batch_once( self, @@ -109,11 +118,15 @@ class SyntheticAcceptanceTransport: else ModelingAction.STOP ) responses[MODELING_RESPONSE_TOPIC] = _modeling_response(action) + if action is ModelingAction.START: + self.start_emitted = True + else: + self.stop_emitted = True return responses for envelope in envelopes: prefix, ordinal_text, message_type = envelope.operation_key.split(":", 2) - assert prefix == "bootstrap" + assert prefix in {"bootstrap", "dialogue"} ordinal = int(ordinal_text) if message_type == "DeviceInfoRequest": session = ( @@ -134,6 +147,36 @@ class SyntheticAcceptanceTransport: responses[topic] = response return responses + def maintain_open_for( + self, + duration_seconds: float, + *, + allowed_response_topics: Collection[str] = (), + ) -> None: + self.maintain_calls.append( + (duration_seconds, tuple(sorted(allowed_response_topics))) + ) + if self.clock is not None: + self.clock.now += duration_seconds + if self.stop_emitted: + self.post_stop_maintain_calls += 1 + elif self.start_emitted: + self.active_maintain_calls += 1 + else: + self.pre_start_maintain_calls += 1 + + def scan_initialization_complete(self, _binding: object) -> bool: + return self.active_maintain_calls >= 2 + + def pre_start_ready(self, _binding: object) -> bool: + return not self.start_emitted + + def standby_complete(self, _binding: object) -> bool: + return self.stop_emitted and self.post_stop_maintain_calls >= 2 + + def validate_bound_status(self, _binding: object) -> None: + return None + def _response_suffix(message_type: str) -> str: values = { @@ -158,17 +201,56 @@ def _checklist(action: ModelingAction) -> PhysicalAcceptanceChecklist: ) -def test_start_acceptance_requires_full_bootstrap_and_consumes_one_short_permit() -> None: - transport = SyntheticAcceptanceTransport() - permit = PhysicalAcceptancePermit(_checklist(ModelingAction.START)) - executor = PhysicalAcceptanceDialogueExecutor(transport, permit) +class FakeClock: + def __init__(self) -> None: + self.now = 100.0 + + def __call__(self) -> float: + return self.now + + +def test_canonical_session_owns_start_active_scan_stop_and_save_boundary() -> None: + clock = FakeClock() + transport = SyntheticAcceptanceTransport(clock) + executor = PhysicalAcceptanceDialogueExecutor(transport) orchestrator = ShadowApplicationBootstrapOrchestrator( ApplicationControlAuthority(openapi_key=APPLICATION_KEY), epoch_seconds=1_752_680_000, timezone_name="Europe/Moscow", ) - binding = executor.run_bootstrap(orchestrator) + with pytest.raises(ApplicationAcceptanceError, match="collapsed bootstrap is disabled"): + executor.run_bootstrap(orchestrator) + + binding = executor.run_connection_stage(orchestrator) + with pytest.raises(ApplicationAcceptanceError, match="not issued by this canonical session"): + executor.run_workspace_entry_stage( + orchestrator, + OperatorDialogueCheckpoint( + event="workspace-entered", + operator_initiated=True, + owner_token=object(), + ), + ) + assert [len(batch) for batch in transport.batches] == [1, 2, 3] + workspace_checks = iter((False, False, True)) + workspace_checkpoint = executor.wait_for_operator_checkpoint( + "workspace-entered", + lambda: next(workspace_checks), + ) + executor.run_workspace_entry_stage( + orchestrator, + workspace_checkpoint, + ) + project_checks = iter((False, False, True)) + project_checkpoint = executor.wait_for_operator_checkpoint( + "project-prompt-opened", + lambda: next(project_checks), + ) + executor.run_project_prompt_stage( + orchestrator, + project_checkpoint, + ) command = ShadowModelingCommand.from_command( encode_modeling_start( CommandHeaderIdentity( @@ -181,23 +263,82 @@ def test_start_acceptance_requires_full_bootstrap_and_consumes_one_short_permit( mount_type=MountType.HANDHELD, ) ) - response = executor.execute_modeling(command) + with pytest.raises(ApplicationAcceptanceError, match="standalone modeling commands"): + executor.execute_modeling(command) + + start_wait_polls = 0 + + def start_confirmed() -> bool: + nonlocal start_wait_polls + start_wait_polls += 1 + return start_wait_polls > 202 + + start_checkpoint = executor.wait_for_operator_checkpoint( + "start-confirmed", + start_confirmed, + ) + # Human/UI time is serviced on the socket and precedes the command permit. + start_permit = PhysicalAcceptancePermit( + _checklist(ModelingAction.START), + ttl_seconds=120.0, + monotonic=clock, + ) + post_start = build_canonical_post_start_observation( + ApplicationControlAuthority(openapi_key=APPLICATION_KEY), + binding, + ) + response = executor.execute_canonical_start( + command, + post_start, + authority=ApplicationControlAuthority(openapi_key=APPLICATION_KEY), + binding=binding, + permit=start_permit, + checkpoint=start_checkpoint, + ) assert response.action is ModelingAction.START - assert [len(batch) for batch in transport.batches] == [1, 2, 3, 1, 3, 1] - assert permit.snapshot()["consumed"] is True + stop_checks = iter((False, False, True)) + executor.maintain_active_until_stop_requested(lambda: next(stop_checks)) + stop_permit = PhysicalAcceptancePermit( + _checklist(ModelingAction.STOP), + ttl_seconds=120.0, + monotonic=clock, + ) + stop_command = ShadowModelingCommand.from_command( + encode_modeling_stop( + CommandHeaderIdentity( + device_id=binding.vendor_device_id, + openapi_key=APPLICATION_KEY, + ) + ) + ) + stop_response = executor.execute_canonical_stop(stop_command, stop_permit) + standby_checks = iter((False, False, True)) + executor.maintain_post_stop_until_standby_confirmed( + lambda: next(standby_checks) + ) + + assert stop_response.action is ModelingAction.STOP + assert [len(batch) for batch in transport.batches] == [1, 2, 3, 1, 3, 1, 1, 2, 1] + assert len(transport.maintain_calls) == 214 + assert set(transport.maintain_calls) == { + (1.0, ("lixel/application/response/modeling_status",)) + } + assert start_permit.snapshot()["consumed"] is True + assert stop_permit.snapshot()["consumed"] is True assert executor.snapshot()["bootstrap_complete"] is True assert executor.snapshot()["command_complete"] is True + assert executor.snapshot()["start_complete"] is True + assert executor.snapshot()["stop_complete"] is True + assert executor.snapshot()["dialogue_stage"] == "standby-confirmed" response_evidence = executor.snapshot()["response_evidence"] assert isinstance(response_evidence, tuple) - assert len(response_evidence) == 10 + assert len(response_evidence) == 13 assert response_evidence[0]["operation_key"] == "bootstrap:1:DeviceInfoRequest" - assert response_evidence[-1]["operation_key"] == "modeling:start" + assert response_evidence[-1]["operation_key"] == "modeling:stop" assert all("payload" not in item for item in response_evidence) assert APPLICATION_KEY not in str(executor.snapshot()) assert VENDOR_DEVICE_ID not in str(executor.snapshot()) - with pytest.raises(ApplicationAcceptanceError, match="already attempted"): - executor.execute_modeling(command) def test_bootstrap_correlation_failure_records_only_redacted_response_evidence() -> None: @@ -220,7 +361,6 @@ def test_bootstrap_correlation_failure_records_only_redacted_response_evidence() executor = PhysicalAcceptanceDialogueExecutor( CorruptDeviceConfigTransport(), - PhysicalAcceptancePermit(_checklist(ModelingAction.START)), ) orchestrator = ShadowApplicationBootstrapOrchestrator( ApplicationControlAuthority(openapi_key=APPLICATION_KEY), @@ -232,7 +372,7 @@ def test_bootstrap_correlation_failure_records_only_redacted_response_evidence() ApplicationAcceptanceError, match=r"ordinal 4 \(DeviceConfigRequest\).*session correlation", ): - executor.run_bootstrap(orchestrator) + executor.run_connection_stage(orchestrator) snapshot = executor.snapshot() failure = snapshot["correlation_failure"] @@ -253,12 +393,9 @@ def test_bootstrap_correlation_failure_records_only_redacted_response_evidence() assert VENDOR_DEVICE_ID not in str(snapshot) -def test_start_cannot_skip_bootstrap_and_stop_uses_a_separate_action_permit() -> None: +def test_standalone_start_and_stop_are_both_disabled() -> None: transport = SyntheticAcceptanceTransport() - start = PhysicalAcceptanceDialogueExecutor( - transport, - PhysicalAcceptancePermit(_checklist(ModelingAction.START)), - ) + start = PhysicalAcceptanceDialogueExecutor(transport) start_command = ShadowModelingCommand.from_command( encode_modeling_start( CommandHeaderIdentity(device_id=VENDOR_DEVICE_ID, openapi_key=APPLICATION_KEY), @@ -268,20 +405,18 @@ def test_start_cannot_skip_bootstrap_and_stop_uses_a_separate_action_permit() -> mount_type=MountType.HANDHELD, ) ) - with pytest.raises(ApplicationAcceptanceError, match="requires.*bootstrap"): + with pytest.raises(ApplicationAcceptanceError, match="standalone modeling commands"): start.execute_modeling(start_command) - stop = PhysicalAcceptanceDialogueExecutor( - transport, - PhysicalAcceptancePermit(_checklist(ModelingAction.STOP)), - ) + stop = PhysicalAcceptanceDialogueExecutor(transport) stop_command = ShadowModelingCommand.from_command( encode_modeling_stop( CommandHeaderIdentity(device_id=VENDOR_DEVICE_ID, openapi_key=APPLICATION_KEY) ) ) - assert stop.execute_modeling(stop_command).action is ModelingAction.STOP - assert transport.batches[-1] == ("modeling:stop",) + with pytest.raises(ApplicationAcceptanceError, match="standalone modeling commands"): + stop.execute_modeling(stop_command) + assert transport.batches == [] def test_acceptance_permit_rejects_implicit_or_expired_authority() -> None: diff --git a/tests/test_xgrids_application_bootstrap.py b/tests/test_xgrids_application_bootstrap.py index 3ef7eba..953f43d 100644 --- a/tests/test_xgrids_application_bootstrap.py +++ b/tests/test_xgrids_application_bootstrap.py @@ -7,6 +7,7 @@ from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import ( ApplicationControlAuthority, LiveDeviceControlBinding, ShadowApplicationBootstrapOrchestrator, + build_canonical_post_start_observation, build_shadow_bootstrap, decode_and_bind_device_info_response, ) @@ -148,6 +149,21 @@ def test_bootstrap_headers_switch_from_unbound_to_live_device_identity() -> None ) +def test_canonical_post_start_observation_preserves_retained_operations_12_to_14() -> None: + plan = build_canonical_post_start_observation(_authority(), _binding()) + + assert [request.ordinal for request in plan.requests] == [12, 13, 14] + assert [request.message_type for request in plan.requests] == [ + "ModelingStatusRequest", + "DeviceInfoRequest", + "ModelingStatusRequest", + ] + assert [request.payload_bytes for request in plan.requests] == [139, 135, 139] + assert [request.response_required for request in plan.requests] == [False, True, True] + assert all(request.phase == "post-start-observation" for request in plan.requests) + assert all(not request.mutates_device for request in plan.requests) + + def test_device_info_response_produces_live_binding_without_a_saved_device_profile() -> None: response = decode_and_bind_device_info_response(_device_info_response(), _authority()) diff --git a/tests/test_xgrids_application_mqtt.py b/tests/test_xgrids_application_mqtt.py index 0b3be30..6daef33 100644 --- a/tests/test_xgrids_application_mqtt.py +++ b/tests/test_xgrids_application_mqtt.py @@ -10,15 +10,28 @@ import pytest from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import ( DEVICE_INFO_RESPONSE_TOPIC, + MODELING_STATUS_RESPONSE_TOPIC, + LiveDeviceControlBinding, ) from k1link.device_plugins.xgrids_k1.protocol.application_mqtt import ( CONTROL_SUBSCRIPTION_GROUPS, + SYSTEM_ERROR_TOPIC, ApplicationCommandOutcomeUnknown, + ApplicationControlDeviceFault, ReviewedApplicationMqttTransport, ) from k1link.device_plugins.xgrids_k1.protocol.application_publish import ( OneShotPublishEnvelope, ) +from k1link.device_plugins.xgrids_k1.protocol.modeling_control import ( + MODELING_STATE_BASE, + SYSTEM_ERROR_STATE_BASE, + SessionState, +) +from k1link.device_plugins.xgrids_k1.protocol.modeling_safety import DEVICE_STATUS_TOPIC + +VENDOR_DEVICE_ID = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee" +DEVICE_SERIAL = "K1SERIAL01" class FakeReasonCode: @@ -35,7 +48,13 @@ class FakeClock: class FakeControlClient: - def __init__(self, *, emit_exchange: bool = True, clock: FakeClock | None = None) -> None: + def __init__( + self, + *, + emit_exchange: bool = True, + clock: FakeClock | None = None, + response_topic: str = DEVICE_INFO_RESPONSE_TOPIC, + ) -> None: self.on_connect: Any = None self.on_subscribe: Any = None self.on_publish: Any = None @@ -43,6 +62,7 @@ class FakeControlClient: self.on_disconnect: Any = None self.emit_exchange = emit_exchange self.clock = clock + self.response_topic = response_topic self.events: deque[tuple[str, object]] = deque() self.connect_calls: list[tuple[str, int, int]] = [] self.subscribe_calls: list[list[tuple[str, int]]] = [] @@ -74,7 +94,7 @@ class FakeControlClient: self.next_mid += 1 if self.emit_exchange: self.events.append(("publish", mid)) - self.events.append(("message", (DEVICE_INFO_RESPONSE_TOPIC, b"response"))) + self.events.append(("message", (self.response_topic, b"response"))) return SimpleNamespace(rc=mqtt.MQTT_ERR_SUCCESS, mid=mid) def loop(self, timeout: float) -> mqtt.MQTTErrorCode: @@ -124,6 +144,70 @@ def _envelope(operation_key: str = "bootstrap:1:DeviceInfoRequest") -> OneShotPu ) +def _modeling_status_envelope() -> OneShotPublishEnvelope: + payload = b"synthetic-reviewed-modeling-status" + return OneShotPublishEnvelope( + operation_key="dialogue:12:ModelingStatusRequest", + topic="lixel/application/request/modeling_status", + payload=payload, + payload_sha256=hashlib.sha256(payload).hexdigest(), + payload_bytes=len(payload), + qos=2, + retain=False, + ) + + +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 _device_status(state: SessionState, *, project_bound: bool, init_ready: bool) -> bytes: + header = _text(4, VENDOR_DEVICE_ID) + fields = [ + _bytes(1, header), + _uint(2, MODELING_STATE_BASE + state), + _text(3, DEVICE_SERIAL), + ] + if project_bound: + fields.append(_text(4, "project-present")) + if init_ready: + fields.append(_uint(6, 1)) + return b"".join(fields) + + +def _system_error(state: SessionState) -> bytes: + return _bytes(15, _uint(1, SYSTEM_ERROR_STATE_BASE + state)) + + +def _binding() -> LiveDeviceControlBinding: + return LiveDeviceControlBinding( + vendor_device_id=VENDOR_DEVICE_ID, + device_serial=DEVICE_SERIAL, + software_version="V3.0.2-20260101-release", + system_version="V3.0.2", + device_model="LixelKity K1", + device_type="K1", + is_activated=True, + ) + + def test_acceptance_transport_rejects_the_k1_access_point_fallback() -> None: with pytest.raises(ValueError, match="not a direct-LAN"): ReviewedApplicationMqttTransport("192.168.56.1") @@ -198,6 +282,96 @@ def test_consumed_operation_key_can_never_be_published_again() -> None: assert len(fake.publish_calls) == 1 +def test_post_start_status_can_be_response_free_and_socket_is_continuously_serviced() -> None: + clock = FakeClock() + fake = FakeControlClient( + clock=clock, + response_topic=MODELING_STATUS_RESPONSE_TOPIC, + ) + transport = ReviewedApplicationMqttTransport( + "192.168.1.20", + client_factory=lambda: cast(mqtt.Client, fake), + monotonic=clock, + ) + transport.open() + + responses = transport.exchange_batch_once( + [_modeling_status_envelope()], + required_response_topics=(), + ) + transport.maintain_open_for( + 2.0, + allowed_response_topics={MODELING_STATUS_RESPONSE_TOPIC}, + ) + + assert responses == {} + assert transport.snapshot().state == "ready" + assert transport.snapshot().qos2_completions == 1 + assert transport.snapshot().ignored_known_responses == 1 + + +def test_control_owner_uses_live_state_gates_and_surfaces_system_error() -> None: + clock = FakeClock() + fake = FakeControlClient(clock=clock) + transport = ReviewedApplicationMqttTransport( + "192.168.1.20", + client_factory=lambda: cast(mqtt.Client, fake), + monotonic=clock, + ) + transport.open() + + fake.events.append( + ( + "message", + ( + DEVICE_STATUS_TOPIC, + _device_status( + SessionState.READY, + project_bound=False, + init_ready=False, + ), + ), + ) + ) + transport.maintain_open_for(1.0) + assert transport.pre_start_ready(_binding()) is True + + fake.events.append( + ( + "message", + ( + DEVICE_STATUS_TOPIC, + _device_status( + SessionState.SCANNING, + project_bound=True, + init_ready=True, + ), + ), + ) + ) + transport.maintain_open_for(1.0) + + assert transport.scan_initialization_complete(_binding()) is True + assert transport.pre_start_ready(_binding()) is False + assert transport.standby_complete(_binding()) is False + + fake.events.append( + ( + "message", + (SYSTEM_ERROR_TOPIC, _system_error(SessionState.ALGORITHM_ERROR)), + ) + ) + transport.maintain_open_for(1.0) + + snapshot = transport.snapshot().as_dict() + assert snapshot["device_status_reports"] == 2 + assert snapshot["system_error_reports"] == 1 + assert snapshot["latest_system_error_code"] == 0x32040133 + assert snapshot["latest_system_error_state"] == "algorithm_error" + with pytest.raises(ApplicationControlDeviceFault, match="algorithm_error"): + transport.scan_initialization_complete(_binding()) + + def test_post_publish_timeout_poisoned_transport_never_retries() -> None: clock = FakeClock() fake = FakeControlClient(emit_exchange=False, clock=clock) diff --git a/tests/test_xgrids_modeling_control_protocol.py b/tests/test_xgrids_modeling_control_protocol.py index 995875f..d891619 100644 --- a/tests/test_xgrids_modeling_control_protocol.py +++ b/tests/test_xgrids_modeling_control_protocol.py @@ -7,6 +7,7 @@ import pytest from k1link.device_plugins.xgrids_k1.protocol.modeling_control import ( MODELING_STATE_BASE, OPENAPI_SUCCESS, + SYSTEM_ERROR_STATE_BASE, CommandHeaderIdentity, ModelingAction, ModelingCommandRejected, @@ -21,6 +22,7 @@ from k1link.device_plugins.xgrids_k1.protocol.modeling_control import ( correlate_modeling_response, decode_device_status_report, decode_modeling_response, + decode_system_error_report, encode_modeling_start, encode_modeling_stop, ) @@ -277,6 +279,22 @@ def test_device_status_maps_base_offset_states_and_preserves_unknown() -> None: decode_device_status_report(_uint(6, 2) + _uint(2, MODELING_STATE_BASE)) +def test_system_error_maps_observed_algorithm_error_without_exposing_identity() -> None: + identity = _identity() + payload = _bytes(1, _header(identity)) + _bytes( + 15, + _uint(1, SYSTEM_ERROR_STATE_BASE + SessionState.ALGORITHM_ERROR), + ) + + report = decode_system_error_report(payload) + + assert report.error_code == 0x3204_0133 + assert report.session_state is SessionState.ALGORITHM_ERROR + assert "synthetic" not in repr(report) + with pytest.raises(ModelingProtocolError, match="no error code"): + decode_system_error_report(_bytes(1, _header(identity))) + + def test_state_machine_never_promotes_status_only_evidence_to_durable_save() -> None: machine = DeviceAcquisitionStateMachine() assert machine.snapshot.phase is AcquisitionPhase.UNOBSERVED