diff --git a/apps/control-station/test/k1SupervisorPresentation.test.mjs b/apps/control-station/test/k1SupervisorPresentation.test.mjs index 4e2ce5d..1f03653 100644 --- a/apps/control-station/test/k1SupervisorPresentation.test.mjs +++ b/apps/control-station/test/k1SupervisorPresentation.test.mjs @@ -6185,6 +6185,20 @@ test("Apply presentation stays correlated through ACK, bootstrap, terminal and c "settling", "ready child success waits for the exact reachable lease projection", ); + assert.equal( + provisioningAttemptViewState(correlated, { + ...initialState, + operations: [operation], + connection_attempt: { + ...attempt, + status: "succeeded", + control_state: "unknown", + safe_next_action: "verify-control-read-only", + }, + }), + "settling", + "the network-success/control-bootstrap handoff keeps the existing loader", + ); assert.equal( provisioningAttemptViewState(correlated, { ...initialState, diff --git a/plugins/xgrids-k1/frontend/src/components/K1ProvisioningPipeline.tsx b/plugins/xgrids-k1/frontend/src/components/K1ProvisioningPipeline.tsx index c9d054d..71654f1 100644 --- a/plugins/xgrids-k1/frontend/src/components/K1ProvisioningPipeline.tsx +++ b/plugins/xgrids-k1/frontend/src/components/K1ProvisioningPipeline.tsx @@ -351,6 +351,18 @@ export function provisioningAttemptViewState( if (attempt.status === "succeeded" && attempt.control_state === "ready") { return "settling"; } + // The network operation can be projected as succeeded immediately before + // its service-owned control bootstrap appears in the next snapshot. Keep + // the click-owned inline progress surface during that exact handoff; a + // terminal bootstrap failure still arrives as failed/control_not_ready. + if ( + attempt.status === "succeeded" + && attempt.phase === "network_applied" + && attempt.control_state === "unknown" + && attempt.safe_next_action === "verify-control-read-only" + ) { + return "settling"; + } if (TERMINAL_PROVISIONING_STATUSES.has(attempt.status)) return "failed"; } if ( diff --git a/src/k1link/device_plugins/xgrids_k1/facade.py b/src/k1link/device_plugins/xgrids_k1/facade.py index a754fdc..9ca0e18 100644 --- a/src/k1link/device_plugins/xgrids_k1/facade.py +++ b/src/k1link/device_plugins/xgrids_k1/facade.py @@ -12885,6 +12885,8 @@ class XgridsK1CompatibilityService: def _validate_active_acquisition_checkpoint_lineage( self, + *, + allow_untrusted_terminal_settlement: bool = False, ) -> _ActiveAcquisitionCheckpointTrustToken | None: """Recompute the current unceased checkpoint/ledger lineage locally. @@ -12892,13 +12894,23 @@ class XgridsK1CompatibilityService: ledger and immutable archive retain its proof material. This method is deliberately read-only: a mismatch revokes local START/resume admission for the process lifetime and never repairs either durable document. + + ``allow_untrusted_terminal_settlement`` is reserved for one narrower + caller which has already durably committed an exact, non-retained + READY/SCAN_OVER reconciliation. It may recompute a token only to cease + the old checkpoint; process-local START/resume trust stays revoked until + that terminal CAS has succeeded. """ - if getattr( + current_trust = getattr( self, "_active_acquisition_checkpoint_trust", "trusted", - ) != "trusted": + ) + if current_trust != "trusted" and not ( + allow_untrusted_terminal_settlement + and current_trust == "unavailable" + ): return None store = self._active_acquisition_checkpoint if store is None: @@ -13023,6 +13035,110 @@ class XgridsK1CompatibilityService: ) return None + def _settle_untrusted_checkpoint_after_verified_standby( + self, + *, + reconciliation_id: str, + reconciled_record: Mapping[str, Any], + ) -> bool: + """Consume a new durable standby proof without reviving START authority. + + A process can reject an ACTIVE checkpoint while an older STOP is still + unresolved. If the same process later appends an exact read-only + READY/SCAN_OVER reconciliation, that durable change is new evidence. + Recompute the lineage only for terminal checkpoint cessation; no K1 + command, network mutation, runtime rehydration, or acquisition authority + is permitted by this path. + """ + + if getattr( + self, + "_active_acquisition_checkpoint_trust", + "trusted", + ) != "unavailable": + return False + store = self._active_acquisition_checkpoint + if store is None: + return False + try: + checkpoint_snapshot = store.snapshot() + ledger_snapshot = self._physical_command_ledger.snapshot() + record = ledger_snapshot.record + reconciliation = ( + record.reconciliations[-1] + if record is not None and record.reconciliations + else None + ) + if not ( + checkpoint_snapshot.status in {"prepared", "active"} + and checkpoint_snapshot.checkpoint is not None + and ledger_snapshot.status == "resolved" + and record is not None + and record.as_dict() == dict(reconciled_record) + and reconciliation is not None + and reconciliation.reconciliation_id == reconciliation_id + and reconciliation.resolution == "physical-standby-observed" + and reconciliation.kind + in { + "ambiguous-outcome", + "prepared-stop-classification", + "resolved-active-cessation", + } + and reconciliation.observation.source + == "explicit-read-only-reconciliation" + and reconciliation.observation.session_state + in {"ready", "scan_over"} + and not reconciliation.observation.project_bound + and not reconciliation.observation.init_ready + and not reconciliation.observation.mqtt_retained + ): + return False + token = self._validate_active_acquisition_checkpoint_lineage( + allow_untrusted_terminal_settlement=True, + ) + if token is None: + return False + settled = self._settle_restart_checkpoint_after_verified_standby( + token=token, + reconciliation_id=reconciliation_id, + reconciled_record=reconciled_record, + ) + if not settled: + return False + committed = store.snapshot().checkpoint + if not ( + committed is not None + and committed.state == "ceased" + and committed.acquisition_id == token.acquisition_id + and committed.original_start_operation_id + == token.root_start_operation_id + ): + return False + with self._lock: + if self._active_acquisition_checkpoint_trust == "unavailable": + self._active_acquisition_checkpoint_trust = "trusted" + self._active_acquisition_checkpoint_reason = None + logger.info( + "untrusted K1 checkpoint closed from new durable standby proof", + extra={ + "event_code": ( + "active_acquisition_checkpoint_untrusted_standby_settled" + ), + "acquisition_id": token.acquisition_id, + "reconciliation_id": reconciliation_id, + "device_write_performed": False, + "automatic_retry": False, + }, + ) + return True + except ( + ActiveAcquisitionRecoveryCheckpointError, + PhysicalCommandLedgerError, + OSError, + ValueError, + ): + return False + def _require_active_acquisition_checkpoint_store( self, ) -> ActiveAcquisitionRecoveryCheckpointStore: @@ -16105,8 +16221,7 @@ class XgridsK1CompatibilityService: else None ) checkpoint_settlement_required = bool( - checkpoint_trust_token is not None - and ledger_snapshot.status == "resolved" + ledger_snapshot.status == "resolved" and record is not None and latest_reconciliation is not None and latest_reconciliation.resolution @@ -16128,14 +16243,24 @@ class XgridsK1CompatibilityService: if checkpoint_settlement_required: assert record is not None assert latest_reconciliation is not None - settled = self._settle_restart_checkpoint_after_verified_standby( - token=checkpoint_trust_token, - reconciliation_id=( - latest_reconciliation.reconciliation_id - ), - reconciled_record=record.as_dict(), - ) - if not settled: + if checkpoint_trust_token is not None: + settled = self._settle_restart_checkpoint_after_verified_standby( + token=checkpoint_trust_token, + reconciliation_id=( + latest_reconciliation.reconciliation_id + ), + reconciled_record=record.as_dict(), + ) + else: + settled = ( + self._settle_untrusted_checkpoint_after_verified_standby( + reconciliation_id=( + latest_reconciliation.reconciliation_id + ), + reconciled_record=record.as_dict(), + ) + ) + if checkpoint_trust_token is not None and not settled: raise ConnectionVerificationError( "Подтверждённый READY не закрыл старую локальную START-сессию", reason_code=( @@ -16369,11 +16494,19 @@ class XgridsK1CompatibilityService: checkpoint_lineage=checkpoint_lineage, ) elif reconciled_record is not None: - settled = self._settle_restart_checkpoint_after_verified_standby( - token=checkpoint_trust_token, - reconciliation_id=reconciliation_id, - reconciled_record=reconciled_record, - ) + if checkpoint_trust_token is not None: + settled = self._settle_restart_checkpoint_after_verified_standby( + token=checkpoint_trust_token, + reconciliation_id=reconciliation_id, + reconciled_record=reconciled_record, + ) + else: + settled = ( + self._settle_untrusted_checkpoint_after_verified_standby( + reconciliation_id=reconciliation_id, + reconciled_record=reconciled_record, + ) + ) if checkpoint_trust_token is not None and not settled: raise ActiveAcquisitionRecoveryCheckpointError( "restart standby checkpoint settlement failed closed" diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index 85b67a4..19c347c 100644 --- a/src/k1link/web/app.py +++ b/src/k1link/web/app.py @@ -535,6 +535,7 @@ _CLOSED_WEBSOCKET_SEND_ERRORS = frozenset( "Unexpected ASGI message 'websocket.send', after sending " "'websocket.close' or response already completed." ), + "Unexpected ASGI message 'websocket.send', after sending 'websocket.close'.", } ) diff --git a/tests/test_web_validation_security.py b/tests/test_web_validation_security.py index f7320cf..a62b547 100644 --- a/tests/test_web_validation_security.py +++ b/tests/test_web_validation_security.py @@ -202,6 +202,10 @@ def test_plugin_expected_state_preserves_its_non_gateway_http_status( "unable to perform operation on ; " "the handler is closed" ), + RuntimeError( + "Unexpected ASGI message 'websocket.send', after sending " + "'websocket.close'." + ), ], ) def test_device_plugin_events_treats_proven_transport_disconnect_as_completion( diff --git a/tests/test_xgrids_acquisition_lifecycle.py b/tests/test_xgrids_acquisition_lifecycle.py index 4d5542e..59e89fc 100644 --- a/tests/test_xgrids_acquisition_lifecycle.py +++ b/tests/test_xgrids_acquisition_lifecycle.py @@ -31256,6 +31256,87 @@ def _persist_same_runtime_prepared_checkpoint_for_resolved_start( ) +def _persist_active_checkpoint_for_resolved_start( + service: XgridsK1CompatibilityService, + *, + connection_mode: facade_module.ConnectionMode, + target_ipv4: str, +) -> None: + """Persist one proven active acquisition without process-local ownership.""" + + _persist_resolved_active_start_for_restart( + service, + connection_mode=connection_mode, + target_ipv4=target_ipv4, + ) + record = service._physical_command_ledger.snapshot().record # noqa: SLF001 + store = service._active_acquisition_checkpoint # noqa: SLF001 + assert record is not None and record.last_status is not None + assert store is not None + connection = record.connection + binding = ActiveAcquisitionRecoveryTransportBinding( + runtime_instance_id=service._snapshot_runtime_id, # noqa: SLF001 + intent_id=connection.intent_id, + transport_ref=connection.transport_ref, + connection_mode=connection.connection_mode, + target_ipv4=connection.target_ipv4, + target_port=connection.target_port, + host_path_epoch=connection.host_path_epoch, + control_session_id=connection.control_session_id, + producer_generation=connection.producer_generation, + logical_device_id="known-k1", + compatibility_profile_id=XGRIDS_K1_COMPATIBILITY_PROFILE_ID, + vendor_device_id_sha256=_RECOVERY_VENDOR_HASH, + device_serial_sha256=_RECOVERY_SERIAL_HASH, + ) + prepared = store.prepare( + transition_id="prepare-active-recovery-start", + predecessor_revision=0, + acquisition_id=record.acquisition_id, + original_start_operation_id=record.operation_id, + start_payload_sha256=record.payload_sha256, + identity=ActiveAcquisitionRecoveryIdentity( + logical_device_id="known-k1", + vendor_device_id_sha256=_RECOVERY_VENDOR_HASH, + device_serial_sha256=_RECOVERY_SERIAL_HASH, + ), + connection=ActiveAcquisitionRecoveryConnection( + transport_ref=connection.transport_ref, + connection_mode=connection.connection_mode, + target_ipv4=connection.target_ipv4, + target_port=connection.target_port, + ), + compatibility_profile_id=XGRIDS_K1_COMPATIBILITY_PROFILE_ID, + project_name="ACTIVE_RECOVERY", + project_name_wire_sha256=active_acquisition_project_name_sha256( + "ACTIVE_RECOVERY" + ), + original_evidence_session_id="evidence-active-recovery", + duration_seconds=None, + requested_streams=("spatial.point-cloud.live", "camera.rgb.live"), + evidence_policy="required", + mount_type="handheld", + gnss_mode="none", + prepared_binding=binding, + ) + store.activate( + transition_id="activate-active-recovery-start", + expected_revision=prepared.revision, + expected_acquisition_id=prepared.acquisition_id, + expected_start_operation_id=prepared.original_start_operation_id, + status_proof=service._checkpoint_status_proof( # noqa: SLF001 + status=record.last_status, + binding=binding, + evidence_session_id="evidence-active-recovery", + ), + physical_proof=service._checkpoint_physical_proof( # noqa: SLF001 + record=record, + binding=binding, + checkpoint=prepared, + ), + ) + + def _persist_resolved_unclassified_stop_for_restart( service: XgridsK1CompatibilityService, *, @@ -32930,6 +33011,125 @@ def test_restart_auto_settles_durable_prepared_resolved_start_standby( assert restarted_runtime.stop_calls == 0 +def test_new_bridge_ready_settles_untrusted_quick_checkpoint_without_command( + tmp_path: Path, +) -> None: + """A new exact READY closes stale Quick state locally in the same process.""" + + service, runtime = service_with_fake_runtime(tmp_path) + _persist_active_checkpoint_for_resolved_start( + service, + connection_mode="quick-connect", + target_ipv4="192.168.56.1", + ) + ledger = service._physical_command_ledger # noqa: SLF001 + start = ledger.snapshot().record + assert start is not None and start.last_status is not None + stop_operation_id = "physical-stop-untrusted-quick-to-bridge" + ledger.prepare( + operation_id=stop_operation_id, + parent_operation_id=start.operation_id, + acquisition_id=start.acquisition_id, + action="stop", + identity=start.identity, + connection=start.connection, + compatibility_profile_id=start.compatibility_profile_id, + payload_sha256="6" * 64, + baseline_status=start.last_status, + ) + ledger.mark_dispatching(stop_operation_id) + ledger.mark_observing( + stop_operation_id, + publish_call_returned=True, + packet_id=42, + ) + ledger.mark_qos2_completed(stop_operation_id, packet_id=42) + ledger.record_application_response( + stop_operation_id, + PhysicalCommandApplicationResponse( + operation_id=stop_operation_id, + action="stop", + control_session_id=start.connection.control_session_id, + host_path_epoch=start.connection.host_path_epoch, + producer_generation=start.connection.producer_generation, + result_code=PHYSICAL_COMMAND_APPLICATION_SUCCESS_CODE, + success=True, + payload_sha256="7" * 64, + observed_at_utc="2026-08-09T12:00:03.000Z", + ), + ) + service._mark_active_acquisition_checkpoint_untrusted( # noqa: SLF001 + trust="unavailable", + reason_code="active-acquisition-recovery-checkpoint-error", + ) + assert service._validate_active_acquisition_checkpoint_lineage() is None # noqa: SLF001 + + coordinator = service._physical_command_coordinator # noqa: SLF001 + coordinator.application_response( + ApplicationMqttResponseEvidence( + operation_key="bootstrap:bridge-ready:DeviceInfoRequest", + response_topic="lixel/application/response/device_info", + payload_sha256="8" * 64, + modeling_action=None, + result_code=None, + success=None, + observed_at_utc="2026-08-09T12:01:00.000Z", + ) + ) + coordinator.bind_control_session( + PhysicalCommandRuntimeBinding( + vendor_device_id_sha256=_RECOVERY_VENDOR_HASH, + device_serial_sha256=_RECOVERY_SERIAL_HASH, + compatibility_profile_id=XGRIDS_K1_COMPATIBILITY_PROFILE_ID, + intent_id="fresh-bridge-intent", + transport_ref="test-ble-transport", + connection_mode="bridge", + target_ipv4="192.168.68.52", + target_port=facade_module.CONTROL_MQTT_PORT, + host_path_epoch=2, + control_session_id="fresh-bridge-control", + producer_generation=2, + ) + ) + coordinator.device_status( + ApplicationMqttDeviceStatusEvidence( + vendor_device_id_sha256=_RECOVERY_VENDOR_HASH, + device_serial_sha256=_RECOVERY_SERIAL_HASH, + session_state="ready", + session_state_code=MODELING_STATE_BASE + 300, + project_bound=False, + project_id_sha256=None, + init_ready=False, + status_message_sha256="9" * 64, + mqtt_retained=False, + observed_at_utc="2026-08-09T12:01:01.000Z", + ) + ) + reconciliation_id = "reconcile-untrusted-quick-to-bridge-ready" + standby_record = coordinator.reconcile_unresolved( + reconciliation_id=reconciliation_id, + ) + + settled = service._settle_untrusted_checkpoint_after_verified_standby( # noqa: SLF001 + reconciliation_id=reconciliation_id, + reconciled_record=standby_record, + ) + + checkpoint_store = service._active_acquisition_checkpoint # noqa: SLF001 + assert checkpoint_store is not None + checkpoint = checkpoint_store.snapshot().checkpoint + assert settled is True + assert checkpoint is not None and checkpoint.state == "ceased" + assert service._active_acquisition_checkpoint_trust == "trusted" # noqa: SLF001 + assert service._active_acquisition_checkpoint_reason is None # noqa: SLF001 + assert standby_record["connection"]["connection_mode"] == "quick-connect" + assert standby_record["reconciliations"][-1]["verified_binding"]["connection"][ + "connection_mode" + ] == "bridge" + assert runtime.start_calls == [] + assert runtime.stop_calls == 0 + + def test_transition_invalid_fences_repeated_verify_until_service_restart( tmp_path: Path, ) -> None: