From 7217244886a29e2ca9eec9eaf566232d42f40099 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Fri, 21 Aug 2026 12:36:50 +0300 Subject: [PATCH] =?UTF-8?q?=D0=A1=D1=82=D0=B0=D0=B1=D0=B8=D0=BB=D1=8C?= =?UTF-8?q?=D0=BD=D0=BE=D0=B5=20=D0=BF=D0=BE=D0=B4=D0=BA=D0=BB=D1=8E=D1=87?= =?UTF-8?q?=D0=B5=D0=BD=D0=B8=D0=B5=20=D0=B2=20=D0=B8=D0=B4=D0=B5=D0=B0?= =?UTF-8?q?=D0=BB=D1=8C=D0=BD=D1=8B=D1=85=20=D1=83=D1=81=D0=BB=D0=BE=D0=B2?= =?UTF-8?q?=D0=B8=D1=8F=D1=85=20K1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/k1link/device_plugins/xgrids_k1/facade.py | 42 +++++ .../xgrids_k1/protocol/application_session.py | 71 +++++-- ...tive_acquisition_checkpoint_integration.py | 21 +++ tests/test_xgrids_application_session.py | 176 ++++++++++++++++++ 4 files changed, 291 insertions(+), 19 deletions(-) diff --git a/src/k1link/device_plugins/xgrids_k1/facade.py b/src/k1link/device_plugins/xgrids_k1/facade.py index 46f5b92..d7a3273 100644 --- a/src/k1link/device_plugins/xgrids_k1/facade.py +++ b/src/k1link/device_plugins/xgrids_k1/facade.py @@ -1791,6 +1791,9 @@ class XgridsK1CompatibilityService: scanning_observer=self._observe_application_scanning_checkpoint, connection_path_validator=self._validate_application_connection_path, connection_binding_validator=self._validate_application_connection_binding, + connection_binding_snapshot_validator=( + self._validate_application_connection_binding_snapshot + ), connection_dispatch_lease=self._acquire_application_dispatch_lease, physical_command_coordinator=self._physical_command_coordinator, ) @@ -3092,6 +3095,18 @@ class XgridsK1CompatibilityService: """Authorize one command only for the exact live DeviceInfo route epoch.""" self._validate_application_connection_path(binding) + self._validate_application_connection_binding_snapshot(binding) + + def _validate_application_connection_binding_snapshot( + self, + binding: ApplicationConnectionBinding, + ) -> None: + """Validate retained control authority without another native route probe. + + This guard is used only around session checkpoints. The MQTT + transport's dispatch lease still calls the full path validator under + the command gate immediately before every physical publish. + """ # The session validates its retained MQTT transport before invoking # this callback. Project a strictly newer remote proof revision into @@ -11526,6 +11541,25 @@ class XgridsK1CompatibilityService: and connection.producer_generation == binding.producer_generation ) + @staticmethod + def _physical_reconciliation_belongs_to_record( + reconciliation: PhysicalCommandReconciliation, + record: PhysicalCommandRecord, + ) -> bool: + """Reject append-only reconciliation history from an older command.""" + + attempt = reconciliation.original_attempt + return bool( + attempt.operation_id == record.operation_id + and attempt.parent_operation_id == record.parent_operation_id + and attempt.acquisition_id == record.acquisition_id + and attempt.action == record.action + and attempt.identity == record.identity + and attempt.connection == record.connection + and attempt.compatibility_profile_id == record.compatibility_profile_id + and attempt.payload_sha256 == record.payload_sha256 + ) + def _active_checkpoint_allows_physical_stop_successor( self, *, @@ -13082,6 +13116,10 @@ class XgridsK1CompatibilityService: record is not None and record.resolution == "not-dispatched" and reconciliation is not None + and self._physical_reconciliation_belongs_to_record( + reconciliation, + record, + ) and reconciliation.kind == "prepared-stop-classification" ) dispatched_then_observed_standby = bool( @@ -13457,6 +13495,10 @@ class XgridsK1CompatibilityService: ) use_reconciliation = bool( reconciliation is not None + and self._physical_reconciliation_belongs_to_record( + reconciliation, + record, + ) and reconciliation.kind in { "ambiguous-outcome", diff --git a/src/k1link/device_plugins/xgrids_k1/protocol/application_session.py b/src/k1link/device_plugins/xgrids_k1/protocol/application_session.py index 7cb49db..69c7b92 100644 --- a/src/k1link/device_plugins/xgrids_k1/protocol/application_session.py +++ b/src/k1link/device_plugins/xgrids_k1/protocol/application_session.py @@ -186,6 +186,7 @@ class InteractiveApplicationControlSession: scanning_observer: ScanningObserver | None = None, connection_path_validator: ConnectionPathValidator | None = None, connection_binding_validator: ConnectionBindingValidator | None = None, + connection_binding_snapshot_validator: ConnectionBindingValidator | None = None, connection_dispatch_lease: ConnectionDispatchLease | None = None, physical_command_coordinator: PhysicalCommandCoordinator | None = None, ) -> None: @@ -199,6 +200,14 @@ class InteractiveApplicationControlSession: # DeviceInfo proof produced by this dialogue. self._connection_path_validator = connection_path_validator or connection_binding_validator self._connection_binding_validator = connection_binding_validator + # Operator/worker checkpoint checks must still reject a revoked + # intent, epoch or control proof, but they must not repeat the native + # CoreWLAN/kernel probe already owned by the transport dispatch lease. + # Falling back to the full validator preserves compatibility for + # integrations which have not split these two proof strengths yet. + self._connection_binding_snapshot_validator = ( + connection_binding_snapshot_validator or connection_binding_validator + ) self._connection_dispatch_lease = connection_dispatch_lease self._physical_command_coordinator = physical_command_coordinator self._scanning_observer_errors = 0 @@ -357,7 +366,7 @@ class InteractiveApplicationControlSession: raise ApplicationAcceptanceError( "read-only inspection has not completed its Verify boundary" ) - self._validate_connection_binding("workspace-entry-operator-preflight") + self._validate_connection_binding_snapshot("workspace-entry-operator-preflight") with self._lock: # The binding check deliberately runs outside the session lock. # Re-check the exact browser checkpoint after it returns so a @@ -438,7 +447,7 @@ class InteractiveApplicationControlSession: expected_state_revision=expected_state_revision, ) self._require_phase_locked("workspace-ready") - self._validate_connection_binding("project-prompt-operator-preflight") + self._validate_connection_binding_snapshot("project-prompt-operator-preflight") with self._lock: self._require_checkpoint_locked( expected_session_generation=expected_session_generation, @@ -463,7 +472,7 @@ class InteractiveApplicationControlSession: ) -> dict[str, object]: confirmation.checklist(ModelingAction.START) if self._physical_command_coordinator is not None: - self._validate_connection_binding("start-prepare-preflight") + self._validate_connection_binding_snapshot("start-prepare-preflight") with self._start_prepare_gate: with self._lock: self._require_checkpoint_locked( @@ -572,7 +581,7 @@ class InteractiveApplicationControlSession: dispatch_admission_deadline_reached ) if self._physical_command_coordinator is not None: - self._validate_connection_binding("stop-prepare-preflight") + self._validate_connection_binding_snapshot("stop-prepare-preflight") self._require_stop_dispatch_deadline_open( dispatch_admission_deadline_reached ) @@ -923,25 +932,25 @@ class InteractiveApplicationControlSession: self._set_phase("scanning") else: if inspection_only: - self._validate_connection_binding( + self._validate_connection_binding_snapshot( "inspection-promotion-pre-dispatch" ) binding = executor.complete_connection_stage( orchestrator, expected_binding=binding, ) - self._validate_connection_binding( + self._validate_connection_binding_snapshot( "inspection-promotion-post-response" ) - self._validate_connection_binding("workspace-entry-pre-dispatch") + self._validate_connection_binding_snapshot("workspace-entry-pre-dispatch") executor.run_workspace_entry_stage( orchestrator, workspace, - dispatch_guard=lambda: self._validate_connection_binding( + dispatch_guard=lambda: self._validate_connection_binding_snapshot( "workspace-entry-dispatch" ), ) - self._validate_connection_binding("workspace-entry-post-response") + self._validate_connection_binding_snapshot("workspace-entry-post-response") self._set_phase("workspace-ready") project = executor.wait_for_operator_checkpoint( @@ -949,15 +958,15 @@ class InteractiveApplicationControlSession: self._project_requested.is_set, ) assert project is not None - self._validate_connection_binding("project-prompt-pre-dispatch") + self._validate_connection_binding_snapshot("project-prompt-pre-dispatch") binding = executor.run_project_prompt_stage( orchestrator, project, - dispatch_guard=lambda: self._validate_connection_binding( + dispatch_guard=lambda: self._validate_connection_binding_snapshot( "project-prompt-dispatch" ), ) - self._validate_connection_binding("project-prompt-post-response") + self._validate_connection_binding_snapshot("project-prompt-post-response") self._set_phase("project-ready") start_checkpoint = executor.wait_for_operator_checkpoint( @@ -969,7 +978,7 @@ class InteractiveApplicationControlSession: start_permit = PhysicalAcceptancePermit( start_confirmation.checklist(ModelingAction.START) ) - self._validate_connection_binding("start-pre-dispatch") + self._validate_connection_binding_snapshot("start-pre-dispatch") self._set_phase("initializing") executor.execute_canonical_start( start_command, @@ -978,11 +987,11 @@ class InteractiveApplicationControlSession: binding=binding, permit=start_permit, checkpoint=start_checkpoint, - dispatch_guard=lambda: self._validate_connection_binding( + dispatch_guard=lambda: self._validate_connection_binding_snapshot( "start-dispatch" ), ) - self._validate_connection_binding("start-post-response") + self._validate_connection_binding_snapshot("start-post-response") if coordinator is not None: coordinator.resolve("start") with self._scanning_transition_gate: @@ -1014,7 +1023,7 @@ class InteractiveApplicationControlSession: self._require_stop_dispatch_deadline_open( stop_dispatch_admission_deadline_reached ) - self._validate_connection_binding("stop-pre-dispatch") + self._validate_connection_binding_snapshot("stop-pre-dispatch") self._require_stop_dispatch_deadline_open( stop_dispatch_admission_deadline_reached ) @@ -1022,12 +1031,14 @@ class InteractiveApplicationControlSession: executor.execute_canonical_stop( stop_command, stop_permit, - dispatch_guard=lambda: self._validate_connection_binding("stop-dispatch"), + dispatch_guard=lambda: self._validate_connection_binding_snapshot( + "stop-dispatch" + ), dispatch_admission_deadline_reached=( stop_dispatch_admission_deadline_reached ), ) - self._validate_connection_binding("stop-post-response") + self._validate_connection_binding_snapshot("stop-post-response") self._set_phase("awaiting-standby-confirmation") executor.maintain_post_stop_until_standby() if coordinator is not None: @@ -1468,11 +1479,33 @@ class InteractiveApplicationControlSession: return True def _validate_connection_binding(self, stage: str) -> None: + with self._lock: + validator = self._connection_binding_validator + self._validate_connection_binding_with_validator(stage, validator) + + def _validate_connection_binding_snapshot(self, stage: str) -> None: + """Check retained authority without another native host-path sample. + + The reviewed MQTT transport still acquires the full connection + dispatch lease immediately before every publish. This lighter guard + is only for the surrounding operator/worker checkpoints where a + second CoreWLAN process would add latency without narrowing the actual + publish race. + """ + + with self._lock: + validator = self._connection_binding_snapshot_validator + self._validate_connection_binding_with_validator(stage, validator) + + def _validate_connection_binding_with_validator( + self, + stage: str, + validator: ConnectionBindingValidator | None, + ) -> None: with self._lock: binding = self._connection_binding live_control_binding = self._live_control_binding transport = self._transport - validator = self._connection_binding_validator if transport is not None: if live_control_binding is None: raise ApplicationConnectionBindingLost( diff --git a/tests/test_xgrids_active_acquisition_checkpoint_integration.py b/tests/test_xgrids_active_acquisition_checkpoint_integration.py index 72673b5..4f3dd9a 100644 --- a/tests/test_xgrids_active_acquisition_checkpoint_integration.py +++ b/tests/test_xgrids_active_acquisition_checkpoint_integration.py @@ -1823,6 +1823,27 @@ def test_chained_stop_facade_cessation_uses_real_ledger_ancestry( assert ancestry is not None assert ancestry.head_parent_operation_id == FIRST_STOP_OPERATION_ID + # Reconciliations are append-only and a successor inherits its parent's + # audit history. Even an otherwise valid standby reconciliation remains + # owned by its original STOP and cannot prove the current STOP. + stale_reconciliation = replace( + first_reconciliation_proof, + resolution="physical-standby-observed", + observation=_status( + first_rebind, + "ready", + source="explicit-read-only-reconciliation", + observed_at_utc="2026-08-13T12:01:05.500Z", + ), + ) + assert stale_reconciliation.original_attempt.operation_id == ( + FIRST_STOP_OPERATION_ID + ) + assert not service._physical_reconciliation_belongs_to_record( # noqa: SLF001 + stale_reconciliation, + final_record, + ) + service._application_control_session.snapshot = lambda: { # type: ignore[method-assign] "verified_control": _verified_control(first_rebind) } diff --git a/tests/test_xgrids_application_session.py b/tests/test_xgrids_application_session.py index 0aca0fb..05c084e 100644 --- a/tests/test_xgrids_application_session.py +++ b/tests/test_xgrids_application_session.py @@ -500,6 +500,182 @@ def test_canonical_stages_require_operator_events_but_device_standby_does_not( ] +def test_session_checkpoints_do_not_repeat_native_dispatch_proof( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Slow host-path proof belongs to the exact publish lease, not UI checkpoints.""" + + class DispatchingExecutor(FakeExecutor): + def _dispatch_one_packet(self) -> None: + guard = self.transport.dispatch_guard + assert guard is not None + release = guard() + try: + self.transport.publish_attempts += 1 + finally: + if callable(release): + release() + + def run_workspace_entry_stage( + self, + orchestrator: object, + checkpoint: object, + *, + dispatch_guard: Callable[[], None] | None = None, + ) -> LiveDeviceControlBinding: + if dispatch_guard is not None: + dispatch_guard() + self._dispatch_one_packet() + return super().run_workspace_entry_stage( + orchestrator, + checkpoint, + dispatch_guard=None, + ) + + def run_project_prompt_stage( + self, + orchestrator: object, + checkpoint: object, + *, + dispatch_guard: Callable[[], None] | None = None, + ) -> LiveDeviceControlBinding: + if dispatch_guard is not None: + dispatch_guard() + self._dispatch_one_packet() + return super().run_project_prompt_stage( + orchestrator, + checkpoint, + dispatch_guard=None, + ) + + def execute_canonical_start(self, *_args: object, **kwargs: object) -> object: + dispatch_guard = kwargs.get("dispatch_guard") + if callable(dispatch_guard): + dispatch_guard() + self._dispatch_one_packet() + return super().execute_canonical_start(*_args, **kwargs) + + def execute_canonical_stop(self, *_args: object, **kwargs: object) -> object: + dispatch_guard = kwargs.get("dispatch_guard") + if callable(dispatch_guard): + dispatch_guard() + self._dispatch_one_packet() + return super().execute_canonical_stop(*_args, **kwargs) + + FakeExecutor.records = [] + monkeypatch.setattr( + session_module, + "PhysicalAcceptanceDialogueExecutor", + DispatchingExecutor, + ) + initial_path_proofs: list[ApplicationConnectionBinding] = [] + full_binding_proofs: list[ApplicationConnectionBinding] = [] + snapshot_proofs: list[ApplicationConnectionBinding] = [] + dispatch_proofs: list[ApplicationConnectionBinding] = [] + dispatch_releases: list[ApplicationConnectionBinding] = [] + + def acquire_dispatch_proof( + binding: ApplicationConnectionBinding, + _deadline_reached: Callable[[], bool] | None, + ) -> Callable[[], None]: + dispatch_proofs.append(binding) + return lambda: dispatch_releases.append(binding) + + transport = FakeTransport("192.168.1.20") + session = InteractiveApplicationControlSession( + FakeAuthorityLoader(), + transport_factory=lambda _host: transport, # type: ignore[arg-type] + connection_path_validator=lambda binding: initial_path_proofs.append(binding), + connection_binding_validator=lambda binding: full_binding_proofs.append(binding), + connection_binding_snapshot_validator=lambda binding: snapshot_proofs.append(binding), + connection_dispatch_lease=acquire_dispatch_proof, + ) + session.open( + host="192.168.1.20", + timezone_name="Europe/Moscow", + confirmation=_confirmation(), + connection_binding=_connection_binding(), + ) + _wait_phase(session, "connection-ready") + session.enter_workspace() + _wait_phase(session, "workspace-ready") + session.open_project_prompt() + _wait_phase(session, "project-ready") + session.request_start(project_name="TEST001", confirmation=_confirmation()) + _wait_phase(session, "scanning") + session.request_stop(confirmation=_confirmation()) + _wait_phase(session, "completed") + + assert initial_path_proofs == [_connection_binding()] + assert full_binding_proofs == [] + assert len(snapshot_proofs) >= 12 + assert dispatch_proofs == [_connection_binding()] * 4 + assert dispatch_releases == dispatch_proofs + + +def test_exact_publish_lease_still_rejects_epoch_drift_after_snapshot_check( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A cheap checkpoint pass can never bypass the final physical route fence.""" + + class DispatchRejectedExecutor(FakeExecutor): + def run_workspace_entry_stage( + self, + orchestrator: object, + checkpoint: object, + *, + dispatch_guard: Callable[[], None] | None = None, + ) -> LiveDeviceControlBinding: + if dispatch_guard is not None: + dispatch_guard() + guard = self.transport.dispatch_guard + assert guard is not None + guard() + return super().run_workspace_entry_stage( + orchestrator, + checkpoint, + dispatch_guard=None, + ) + + FakeExecutor.records = [] + monkeypatch.setattr( + session_module, + "PhysicalAcceptanceDialogueExecutor", + DispatchRejectedExecutor, + ) + snapshot_proofs: list[ApplicationConnectionBinding] = [] + + def reject_dispatch( + _binding: ApplicationConnectionBinding, + _deadline_reached: Callable[[], bool] | None, + ) -> Callable[[], None]: + raise ApplicationConnectionBindingLost("test host-path epoch changed") + + session = InteractiveApplicationControlSession( + FakeAuthorityLoader(), + transport_factory=lambda host: FakeTransport(host), # type: ignore[arg-type] + connection_path_validator=lambda _binding: True, + connection_binding_validator=lambda _binding: True, + connection_binding_snapshot_validator=lambda binding: snapshot_proofs.append(binding), + connection_dispatch_lease=reject_dispatch, + ) + session.open( + host="192.168.1.20", + timezone_name="Europe/Moscow", + confirmation=_confirmation(), + connection_binding=_connection_binding(), + ) + _wait_phase(session, "connection-ready") + session.enter_workspace() + failed = _wait_phase(session, "failed") + + assert snapshot_proofs + assert "workspace:7" not in FakeExecutor.records + assert failed["failure"]["reason_code"] == ( # type: ignore[index] + ApplicationConnectionBindingLost.reason_code + ) + + def test_read_only_device_info_open_does_not_require_physical_acceptance( monkeypatch: pytest.MonkeyPatch, ) -> None: