chore(k1): checkpoint connection recovery work

2026-08-21 08:12:22 +03:00
parent a3138f3d71
commit c7843a3c7e
22 changed files with 969 additions and 293 deletions
@@ -1989,9 +1989,11 @@ class ActiveAcquisitionRecoveryCheckpointStore:
                raise ActiveAcquisitionRecoveryCheckpointTransitionError(
                    "only a prepared or active checkpoint can cease"
                )
-            _require_physical_lineage_base(current, physical_proof)
            transport_revision = current.transport_revision
            current_binding = current.current_binding
+            gap_started_at = current.last_gap_started_at_utc
+            gap_started_revision = current.last_gap_started_transport_revision
+            gap_failed_binding = current.last_gap_failed_binding
            gap_recovered_at = current.last_gap_recovered_at_utc
            gap_recovered_revision = current.last_gap_recovered_transport_revision
            prepared_resolution: ActiveAcquisitionRecoveryPhysicalLineageProof | None = None
@@ -2003,7 +2005,21 @@ class ActiveAcquisitionRecoveryCheckpointStore:
                )
                prepared_resolution = physical_proof
                current_binding = physical_proof.binding
+                if (
+                    physical_proof.binding.target_ipv4
+                    != current.connection.target_ipv4
+                ):
+                    transport_revision = _next_transport_revision(
+                        current.transport_revision
+                    )
+                    gap_started_at = current.updated_at_utc
+                    gap_started_revision = current.transport_revision
+                    gap_failed_binding = current.prepared_binding
+                    assert status_proof is not None
+                    gap_recovered_at = status_proof.observed_at_utc
+                    gap_recovered_revision = transport_revision
            else:
+                _require_physical_lineage_base(current, physical_proof)
                assert status_proof is not None
                _require_active_cessation(
                    current,
@@ -2054,6 +2070,9 @@ class ActiveAcquisitionRecoveryCheckpointStore:
                ),
                updated_at_utc=now,
                ceased_at_utc=ceased_at,
+                last_gap_started_at_utc=gap_started_at,
+                last_gap_started_transport_revision=gap_started_revision,
+                last_gap_failed_binding=gap_failed_binding,
                last_gap_recovered_at_utc=gap_recovered_at,
                last_gap_recovered_transport_revision=gap_recovered_revision,
                cessation_status_proof=status_proof,
@@ -2594,6 +2613,7 @@ def _require_binding_matches_checkpoint_values(
    connection: ActiveAcquisitionRecoveryConnection,
    compatibility_profile_id: str,
    binding: ActiveAcquisitionRecoveryTransportBinding,
+    allow_target_ipv4_change: bool = False,
 ) -> None:
    if (
        binding.logical_device_id != identity.logical_device_id
@@ -2602,7 +2622,10 @@ def _require_binding_matches_checkpoint_values(
        or binding.compatibility_profile_id != compatibility_profile_id
        or binding.transport_ref != connection.transport_ref
        or binding.connection_mode != connection.connection_mode
-        or binding.target_ipv4 != connection.target_ipv4
+        or (
+            not allow_target_ipv4_change
+            and binding.target_ipv4 != connection.target_ipv4
+        )
        or binding.target_port != connection.target_port
    ):
        raise ActiveAcquisitionRecoveryCheckpointTransitionError(
@@ -2619,6 +2642,11 @@ def _require_status_binding(
        connection=checkpoint.connection,
        compatibility_profile_id=checkpoint.compatibility_profile_id,
        binding=proof.binding,
+        allow_target_ipv4_change=(
+            checkpoint.state != "prepared"
+            and checkpoint.last_gap_recovered_at_utc is not None
+            and proof.binding == checkpoint.current_binding
+        ),
    )


@@ -2654,6 +2682,8 @@ def _require_gap_cessation_evidence_session(
 def _require_physical_lineage_base(
    checkpoint: ActiveAcquisitionRecoveryCheckpoint,
    proof: ActiveAcquisitionRecoveryPhysicalLineageProof,
+    *,
+    allow_target_ipv4_change: bool = False,
 ) -> None:
    if (
        proof.acquisition_id != checkpoint.acquisition_id
@@ -2668,6 +2698,14 @@ def _require_physical_lineage_base(
        connection=checkpoint.connection,
        compatibility_profile_id=checkpoint.compatibility_profile_id,
        binding=proof.binding,
+        allow_target_ipv4_change=(
+            allow_target_ipv4_change
+            or (
+                checkpoint.state != "prepared"
+                and checkpoint.last_gap_recovered_at_utc is not None
+                and proof.binding == checkpoint.current_binding
+            )
+        ),
    )


@@ -3103,6 +3141,14 @@ def _require_prepared_cessation(
    status_proof: ActiveAcquisitionRecoveryStatusProof | None,
    physical_proof: ActiveAcquisitionRecoveryPhysicalLineageProof,
 ) -> None:
+    target_ipv4_changed = (
+        physical_proof.binding.target_ipv4 != checkpoint.connection.target_ipv4
+    )
+    _require_physical_lineage_base(
+        checkpoint,
+        physical_proof,
+        allow_target_ipv4_change=target_ipv4_changed,
+    )
    if (
        physical_proof.operation_id != checkpoint.original_start_operation_id
        or physical_proof.action != "start"
@@ -3138,7 +3184,26 @@ def _require_prepared_cessation(
        raise ActiveAcquisitionRecoveryCheckpointTransitionError(
            "prepared terminal reconciliation requires fresh READY/SCAN_OVER"
        )
-    _require_status_binding(checkpoint, status_proof)
+    _require_binding_matches_checkpoint_values(
+        identity=checkpoint.identity,
+        connection=checkpoint.connection,
+        compatibility_profile_id=checkpoint.compatibility_profile_id,
+        binding=status_proof.binding,
+        allow_target_ipv4_change=target_ipv4_changed,
+    )
+    if target_ipv4_changed and (
+        status_proof.source != "explicit-read-only-reconciliation"
+        or status_proof.binding.runtime_instance_id
+        == checkpoint.prepared_binding.runtime_instance_id
+        or status_proof.binding.control_session_id
+        == checkpoint.prepared_binding.control_session_id
+        or status_proof.evidence_session_id
+        == checkpoint.original_evidence_session_id
+    ):
+        raise ActiveAcquisitionRecoveryCheckpointTransitionError(
+            "prepared target change requires fresh DeviceInfo-bound runtime, "
+            "control and evidence sessions"
+        )
    _require_matching_observed_proofs(
        status_proof=status_proof,
        physical_proof=physical_proof,
@@ -3331,16 +3396,13 @@ def _require_active_reconciled_standby_shape(
        == failed_evidence_session_id
        or (
            same_runtime
-            and (
-                recovery_binding.host_path_epoch == failed_binding.host_path_epoch
-                or recovery_binding.producer_generation
-                == failed_binding.producer_generation
-            )
+            and recovery_binding.producer_generation
+            == failed_binding.producer_generation
        )
    ):
        raise ActiveAcquisitionRecoveryCheckpointTransitionError(
            "standby settlement requires a new control/evidence boundary; "
-            "same-runtime recovery also requires a new host path and producer generation"
+            "same-runtime recovery also requires a new producer generation"
        )

    gap_started = _validated_timestamp(
@@ -4565,19 +4627,29 @@ def _validate_checkpoint_semantics(
    _validate_evidence_policy(checkpoint.evidence_policy)
    _validate_mount_type(checkpoint.mount_type)
    _validate_gnss_mode(checkpoint.gnss_mode)
-    for binding in (checkpoint.prepared_binding, checkpoint.current_binding):
-        _require_binding_matches_checkpoint_values(
-            identity=checkpoint.identity,
-            connection=checkpoint.connection,
-            compatibility_profile_id=checkpoint.compatibility_profile_id,
-            binding=binding,
-        )
+    _require_binding_matches_checkpoint_values(
+        identity=checkpoint.identity,
+        connection=checkpoint.connection,
+        compatibility_profile_id=checkpoint.compatibility_profile_id,
+        binding=checkpoint.prepared_binding,
+    )
+    _require_binding_matches_checkpoint_values(
+        identity=checkpoint.identity,
+        connection=checkpoint.connection,
+        compatibility_profile_id=checkpoint.compatibility_profile_id,
+        binding=checkpoint.current_binding,
+        allow_target_ipv4_change=(
+            checkpoint.state != "prepared"
+            and checkpoint.last_gap_recovered_at_utc is not None
+        ),
+    )
    if checkpoint.last_gap_failed_binding is not None:
        _require_binding_matches_checkpoint_values(
            identity=checkpoint.identity,
            connection=checkpoint.connection,
            compatibility_profile_id=checkpoint.compatibility_profile_id,
            binding=checkpoint.last_gap_failed_binding,
+            allow_target_ipv4_change=(checkpoint.state != "prepared"),
        )
    _canonical_timestamp_value(checkpoint.created_at_utc, field_name="created_at_utc")
    _canonical_timestamp_value(checkpoint.updated_at_utc, field_name="updated_at_utc")
@@ -3555,6 +3555,66 @@ class XgridsK1CompatibilityService:
            self._application_control_session.snapshot()
        )

+    def _retire_verified_inspection_for_operator_dialogue(
+        self,
+        inspection: Mapping[str, Any],
+    ) -> None:
+        """Close Verify's ordinal-1 socket before a canonical operator dialogue.
+
+        Verify is intentionally read-only and may remain open while the Park UI
+        waits for the next explicit action. The captured LixelGO dialogue does
+        not resume that old socket: ordinals 1-6 are one contiguous connection
+        stage on a fresh socket. Retiring the inspection here is local-only;
+        it cannot publish START, STOP, Wi-Fi credentials, or another K1 request.
+        """
+
+        generation = inspection.get("session_generation")
+        revision = inspection.get("state_revision")
+        if not isinstance(generation, int) or not isinstance(revision, int):
+            raise ApplicationAcceptanceError(
+                "read-only inspection checkpoint is unavailable"
+            )
+        if not (
+            inspection.get("state") == "connection-ready"
+            and inspection.get("inspection_only") is True
+        ):
+            raise ApplicationAcceptanceError(
+                "only a connection-ready inspection may be replaced"
+            )
+
+        self._application_control_session.close_prestart(
+            expected_session_generation=generation,
+            expected_state_revision=revision,
+        )
+        deadline = time.monotonic() + CONTROL_LOCAL_RETIREMENT_TIMEOUT_SECONDS
+        while True:
+            terminal = dict(self._application_control_session.snapshot())
+            if terminal.get("can_open") is True:
+                break
+            if time.monotonic() >= deadline:
+                # Bounded local socket teardown only. ``close`` never invents
+                # a physical STOP for a pre-START session.
+                self._application_control_session.close()
+                terminal = dict(self._application_control_session.snapshot())
+                if terminal.get("can_open") is not True:
+                    raise ApplicationAcceptanceError(
+                        "read-only inspection worker is still retiring"
+                    )
+                break
+            time.sleep(0.05)
+
+        self._retire_application_control_for_network_change(
+            allow_terminal_failure=True,
+        )
+        logger.info(
+            "K1 read-only inspection retired before canonical operator dialogue",
+            extra={
+                "event_code": "k1_control_inspection_retired_before_operator_dialogue",
+                "device_write_performed": False,
+                "automatic_retry": False,
+            },
+        )
+
    def _retry_pending_local_control_retirement(self) -> bool:
        """Finish local socket retirement after its worker actually exits."""

@@ -4400,32 +4460,6 @@ class XgridsK1CompatibilityService:
                        allow_receiver_rehydrate=False,
                    )
                )
-                if (
-                    physical_reconciliation.get("resolution")
-                    != "physical-active-observed"
-                ):
-                    inspected = self._application_control_session.snapshot()
-                    if (
-                        inspected.get("state") == "connection-ready"
-                        and inspected.get("inspection_only") is True
-                        and inspected.get("inspection_promotion_allowed") is not True
-                    ):
-                        generation = inspected.get("session_generation")
-                        revision = inspected.get("state_revision")
-                        if not isinstance(generation, int) or not isinstance(
-                            revision,
-                            int,
-                        ):
-                            raise ConnectionVerificationError(
-                                "read-only continuation lost its inspection checkpoint",
-                                reason_code=(
-                                    "control-bootstrap-inspection-checkpoint-invalid"
-                                ),
-                            )
-                        self._application_control_session.release_inspection_for_operator_dialogue(
-                            expected_session_generation=generation,
-                            expected_state_revision=revision,
-                        )
                if (
                    physical_reconciliation.get("resolution")
                    == "physical-active-observed"
@@ -14816,23 +14850,6 @@ class XgridsK1CompatibilityService:
                physical_reconciliation = await self._reconcile_physical_command_after_verify_owned(
                    verify_operation_id=verify_operation_id,
                )
-                verified_session = dict(self._application_control_session.snapshot())
-                if (
-                    verified_session.get("state") == "connection-ready"
-                    and verified_session.get("inspection_only") is True
-                    and verified_session.get("inspection_promotion_allowed") is not True
-                ):
-                    generation = verified_session.get("session_generation")
-                    revision = verified_session.get("state_revision")
-                    if not isinstance(generation, int) or not isinstance(revision, int):
-                        raise ConnectionVerificationError(
-                            "Verify не получил точный checkpoint inspection-сессии",
-                            reason_code="connection-verify-control-checkpoint-invalid",
-                        )
-                    self._application_control_session.release_inspection_for_operator_dialogue(
-                        expected_session_generation=generation,
-                        expected_state_revision=revision,
-                    )
                if provisional_topology is not None:
                    control_stage = "semantic-topology-commit"
                    self._commit_provisional_fresh_bridge_topology(provisional_topology)
@@ -15023,6 +15040,59 @@ class XgridsK1CompatibilityService:
                or resolved_active_recovery_required
                or resolved_scan_over_recovery_required
            ):
+                # The physical-ledger reconciliation may have committed before
+                # an older process failed to persist the matching checkpoint
+                # cessation.  A later explicit Verify must close that durable
+                # fsync gap from the immutable latest READY/SCAN_OVER proof;
+                # otherwise the UI can look connected while the next START is
+                # still fail-closed by a stale PREPARED checkpoint.  This is a
+                # local projection only and never publishes a K1 command.
+                ledger_snapshot = self._physical_command_ledger.snapshot()
+                record = ledger_snapshot.record
+                latest_reconciliation = (
+                    record.reconciliations[-1]
+                    if record is not None and record.reconciliations
+                    else None
+                )
+                checkpoint_settlement_required = bool(
+                    checkpoint_trust_token is not None
+                    and ledger_snapshot.status == "resolved"
+                    and record is not None
+                    and latest_reconciliation is not None
+                    and latest_reconciliation.resolution
+                    == "physical-standby-observed"
+                    and latest_reconciliation.kind
+                    in {
+                        "ambiguous-outcome",
+                        "prepared-stop-classification",
+                        "resolved-active-cessation",
+                    }
+                    and latest_reconciliation.observation.source
+                    == "explicit-read-only-reconciliation"
+                    and latest_reconciliation.observation.session_state
+                    in {"ready", "scan_over"}
+                    and not latest_reconciliation.observation.project_bound
+                    and not latest_reconciliation.observation.init_ready
+                    and not latest_reconciliation.observation.mqtt_retained
+                )
+                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:
+                        raise ConnectionVerificationError(
+                            "Подтверждённый READY не закрыл старую локальную START-сессию",
+                            reason_code=(
+                                self._active_acquisition_checkpoint_reason
+                                or "restart-standby-checkpoint-settlement-failed"
+                            ),
+                        )
                return {
                    "performed": False,
                    "resolution": None,
@@ -19052,6 +19122,14 @@ class XgridsK1CompatibilityService:
            if acquisition is not None and acquisition.state not in TERMINAL_ACQUISITION_STATES:
                raise RuntimeError("control-сессия должна быть открыта до подготовки acquisition")
            existing_control = self._application_control_session.snapshot()
+            if (
+                existing_control.get("state") == "connection-ready"
+                and existing_control.get("inspection_only") is True
+            ):
+                self._retire_verified_inspection_for_operator_dialogue(
+                    existing_control,
+                )
+                existing_control = self._application_control_session.snapshot()
            if existing_control.get("state") in {
                "connection-ready",
                "workspace-ready",
@@ -265,7 +265,8 @@ class InteractiveApplicationControlSession:
        # Opening the MQTT dialogue begins with only the read-only DeviceInfo
        # bootstrap.  Durable identity admission must accept that proof before
        # a normal session may continue its pre-START preparation; inspection
-        # sessions remain read-only until explicitly promoted.  Physical
+        # sessions remain read-only and must retire before a normal dialogue.
+        # Physical
        # acceptance belongs to the exact START/STOP dispatch boundaries below;
        # a connection intent must never fabricate those confirmations merely
        # to prove that the selected K1 is ready.
@@ -375,27 +376,6 @@ class InteractiveApplicationControlSession:
            self._workspace_requested.set()
        return self.snapshot()

-    def release_inspection_for_operator_dialogue(
-        self,
-        *,
-        expected_session_generation: int,
-        expected_state_revision: int,
-    ) -> dict[str, object]:
-        """End Verify's read-only boundary without sending another request."""
-
-        with self._lock:
-            self._require_checkpoint_locked(
-                expected_session_generation=expected_session_generation,
-                expected_state_revision=expected_state_revision,
-            )
-            self._require_phase_locked("connection-ready")
-            if not self._inspection_only:
-                return self.snapshot()
-            if not self._inspection_promotion_allowed:
-                self._inspection_promotion_allowed = True
-                self._state_revision += 1
-        return self.snapshot()
-
    def validate_connection_binding(self) -> None:
        """Fail closed when the DeviceInfo-bound route lost command authority."""