diff --git a/src/k1link/device_plugins/xgrids_k1/active_acquisition_recovery_checkpoint.py b/src/k1link/device_plugins/xgrids_k1/active_acquisition_recovery_checkpoint.py index 2f1a344..4d7325a 100644 --- a/src/k1link/device_plugins/xgrids_k1/active_acquisition_recovery_checkpoint.py +++ b/src/k1link/device_plugins/xgrids_k1/active_acquisition_recovery_checkpoint.py @@ -3217,11 +3217,22 @@ def _require_active_cessation_shape( and physical_proof.reconciliation_resolution == "physical-standby-observed" and physical_proof.stop_fence == "definitely-not-dispatched" ) + reconciled_ambiguous_stop_standby = bool( + physical_proof.action == "stop" + and physical_proof.operation_id != checkpoint.original_start_operation_id + and _stop_lineage_reaches_checkpoint_start(checkpoint, physical_proof) + and physical_proof.resolution == "physical-standby-observed" + and not physical_proof.composite_complete + and physical_proof.reconciliation_kind == "ambiguous-outcome" + and physical_proof.reconciliation_resolution == "physical-standby-observed" + and physical_proof.stop_fence == "none" + ) if not ( ordinary_stop or reconciled_start_standby or reconciled_ambiguous_start_standby or reconciled_undispatched_stop_standby + or reconciled_ambiguous_stop_standby ): raise ActiveAcquisitionRecoveryCheckpointTransitionError( "active cessation lacks terminal physical standby lineage" @@ -3269,7 +3280,11 @@ def _require_active_reconciled_standby_shape( ) if not ( cessation_physical_proof.reconciliation_kind - in {"prepared-stop-classification", "resolved-active-cessation"} + in { + "ambiguous-outcome", + "prepared-stop-classification", + "resolved-active-cessation", + } and cessation_physical_proof.reconciliation_resolution == "physical-standby-observed" and cessation_physical_proof.effective_resolution diff --git a/src/k1link/device_plugins/xgrids_k1/facade.py b/src/k1link/device_plugins/xgrids_k1/facade.py index ba69ae4..a8dd04b 100644 --- a/src/k1link/device_plugins/xgrids_k1/facade.py +++ b/src/k1link/device_plugins/xgrids_k1/facade.py @@ -11498,11 +11498,38 @@ class XgridsK1CompatibilityService: == "explicit-read-only-reconciliation" and not reconciliation.observation.mqtt_retained ) + reconciled_ambiguous_terminal_stop = bool( + record.stage == "resolved" + and record.resolution == "physical-standby-observed" + and reconciliation is not None + and reconciliation.kind == "ambiguous-outcome" + and reconciliation.resolution == "physical-standby-observed" + and reconciliation.original_attempt.operation_id == record.operation_id + and reconciliation.original_attempt.acquisition_id == record.acquisition_id + and reconciliation.original_attempt.action == "stop" + and reconciliation.original_attempt.stage == "observing" + and reconciliation.original_attempt.resolution is None + and reconciliation.original_attempt.publish_call_returned is True + and reconciliation.original_attempt.packet_id is not None + and reconciliation.original_attempt.qos2_completed + and reconciliation.original_attempt.application_response is not None + and reconciliation.original_attempt.application_response.success + 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 + ) if not ( persisted is not None and status is not None and record.action == "stop" - and (unresolved_stop or classified_terminal_stop) + and ( + unresolved_stop + or classified_terminal_stop + or reconciled_ambiguous_terminal_stop + ) and record.acquisition_id == checkpoint.acquisition_id and record.identity.vendor_device_id_sha256 == checkpoint.identity.vendor_device_id_sha256 @@ -12979,7 +13006,7 @@ class XgridsK1CompatibilityService: self, token: _ActiveAcquisitionCheckpointTrustToken | None, ) -> bool: - """Finish the exact fsync-after-ledger crash gap without device I/O.""" + """Finish an exact STOP-ledger/checkpoint fsync gap without device I/O.""" if token is None or token.checkpoint_state != "active": return False @@ -12990,14 +13017,35 @@ class XgridsK1CompatibilityService: if record is not None and record.reconciliations else None ) + definitely_undispatched = bool( + record is not None + and record.resolution == "not-dispatched" + and reconciliation is not None + and reconciliation.kind == "prepared-stop-classification" + ) + dispatched_then_observed_standby = bool( + record is not None + and record.resolution == "physical-standby-observed" + and reconciliation is not None + and reconciliation.kind == "ambiguous-outcome" + and reconciliation.original_attempt.operation_id == record.operation_id + and reconciliation.original_attempt.acquisition_id == record.acquisition_id + and reconciliation.original_attempt.action == "stop" + and reconciliation.original_attempt.stage == "observing" + and reconciliation.original_attempt.resolution is None + and reconciliation.original_attempt.publish_call_returned is True + and reconciliation.original_attempt.packet_id is not None + and reconciliation.original_attempt.qos2_completed + and reconciliation.original_attempt.application_response is not None + and reconciliation.original_attempt.application_response.success + ) if not ( ledger_snapshot.status == "resolved" and record is not None and record.action == "stop" and record.stage == "resolved" - and record.resolution == "not-dispatched" and reconciliation is not None - and reconciliation.kind == "prepared-stop-classification" + and (definitely_undispatched or dispatched_then_observed_standby) and reconciliation.resolution == "physical-standby-observed" and reconciliation.observation.source == "explicit-read-only-reconciliation" diff --git a/tests/test_xgrids_active_acquisition_checkpoint_integration.py b/tests/test_xgrids_active_acquisition_checkpoint_integration.py index e4d84ef..87944d2 100644 --- a/tests/test_xgrids_active_acquisition_checkpoint_integration.py +++ b/tests/test_xgrids_active_acquisition_checkpoint_integration.py @@ -26,6 +26,7 @@ from k1link.device_plugins.xgrids_k1.connection_supervisor import ( ) from k1link.device_plugins.xgrids_k1.facade import ( XGRIDS_K1_COMPATIBILITY_PROFILE_ID, + ConnectionVerificationError, OperatorPresenceRequest, StopAcquisitionRequest, XgridsK1CompatibilityService, @@ -454,6 +455,143 @@ def test_repeated_reset_reopen_fresh_ready_ceases_old_active_checkpoint( assert service._active_acquisition_checkpoint_trust == "trusted" # noqa: SLF001 +def test_restart_ready_after_dispatched_stop_ceases_active_checkpoint( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Fresh READY settles a dispatched STOP whose outcome crossed restart.""" + + service = _service(tmp_path, monkeypatch) + original = _connection( + control_session_id="checkpoint-dispatched-stop-original-control", + host_path_epoch=1, + producer_generation=1, + ) + _prepare_and_activate_checkpoint(service, original) + ledger = service._physical_command_ledger # noqa: SLF001 + identity = PhysicalCommandIdentity( + vendor_device_id_sha256=VENDOR_SHA256, + device_serial_sha256=SERIAL_SHA256, + ) + ledger.prepare( + operation_id=FIRST_STOP_OPERATION_ID, + parent_operation_id=START_OPERATION_ID, + acquisition_id=ACQUISITION_ID, + action="stop", + identity=identity, + connection=original, + compatibility_profile_id=XGRIDS_K1_COMPATIBILITY_PROFILE_ID, + payload_sha256="2" * 64, + baseline_status=_status( + original, + "scanning", + observed_at_utc="2026-08-13T12:01:00.000Z", + ), + ) + ledger.mark_dispatching(FIRST_STOP_OPERATION_ID) + ledger.mark_observing( + FIRST_STOP_OPERATION_ID, + publish_call_returned=True, + packet_id=42, + ) + ledger.mark_qos2_completed(FIRST_STOP_OPERATION_ID, packet_id=42) + ledger.record_application_response( + FIRST_STOP_OPERATION_ID, + PhysicalCommandApplicationResponse( + operation_id=FIRST_STOP_OPERATION_ID, + action="stop", + control_session_id=original.control_session_id, + host_path_epoch=original.host_path_epoch, + producer_generation=original.producer_generation, + result_code=PHYSICAL_COMMAND_APPLICATION_SUCCESS_CODE, + success=True, + payload_sha256="3" * 64, + observed_at_utc="2026-08-13T12:01:01.000Z", + ), + ) + service._snapshot_runtime_id = ( # noqa: SLF001 + "snapshot-runtime-checkpoint-dispatched-stop-successor" + ) + restarted = _connection( + control_session_id="checkpoint-dispatched-stop-restarted-control", + host_path_epoch=2, + producer_generation=2, + ) + coordinator = service._physical_command_coordinator # noqa: SLF001 + coordinator.application_response( + ApplicationMqttResponseEvidence( + operation_key="bootstrap:dispatched-stop-ready:DeviceInfoRequest", + response_topic="lixel/application/response/device_info", + payload_sha256="4" * 64, + modeling_action=None, + result_code=None, + success=None, + observed_at_utc="2026-08-13T12:02:00.000Z", + ) + ) + coordinator.bind_control_session(_runtime_binding(restarted)) + coordinator.device_status( + ApplicationMqttDeviceStatusEvidence( + vendor_device_id_sha256=VENDOR_SHA256, + device_serial_sha256=SERIAL_SHA256, + session_state="ready", + session_state_code=MODELING_STATE_BASE + 300, + project_bound=False, + project_id_sha256=None, + init_ready=False, + status_message_sha256="5" * 64, + mqtt_retained=False, + observed_at_utc="2026-08-13T12:02:01.000Z", + ) + ) + control_snapshot = { + "state": "connection-ready", + "session_generation": 2, + "state_revision": 2, + "verified_control": _verified_control(restarted), + } + monkeypatch.setattr( + service._application_control_session, # noqa: SLF001 + "snapshot", + lambda: dict(control_snapshot), + ) + monkeypatch.setattr( + service._application_control_session, # noqa: SLF001 + "validate_physical_reconciliation_binding", + lambda: None, + ) + + monkeypatch.setattr( + service, + "_settle_restart_checkpoint_after_verified_standby", + lambda **_kwargs: False, + ) + with pytest.raises(ConnectionVerificationError): + asyncio.run( + service._reconcile_physical_command_after_verify_owned( # noqa: SLF001 + verify_operation_id="checkpoint-dispatched-stop-ready-verify", + allow_receiver_rehydrate=False, + ) + ) + + record = ledger.snapshot().record + assert record is not None + assert record.action == "stop" + assert record.resolution == "physical-standby-observed" + assert record.reconciliations[-1].kind == "ambiguous-outcome" + checkpoint = ActiveAcquisitionRecoveryCheckpointStore(tmp_path).snapshot().checkpoint + assert checkpoint is not None + assert checkpoint.state == "active" + + # Simulate the exact live process boundary: the physical ledger fsync is + # durable, while checkpoint cessation did not commit in the prior process. + restarted_service = _service(tmp_path, monkeypatch) + assert restarted_service._active_acquisition_checkpoint_reason is None # noqa: SLF001 + checkpoint = ActiveAcquisitionRecoveryCheckpointStore(tmp_path).snapshot().checkpoint + assert checkpoint is not None + assert checkpoint.state == "ceased" + + def _published_point_cloud(sequence: int = 41) -> DecodedPointCloudView: return DecodedPointCloudView( context=ConsumerFrameContext( diff --git a/tests/test_xgrids_active_acquisition_recovery_checkpoint.py b/tests/test_xgrids_active_acquisition_recovery_checkpoint.py index d524b75..daa255c 100644 --- a/tests/test_xgrids_active_acquisition_recovery_checkpoint.py +++ b/tests/test_xgrids_active_acquisition_recovery_checkpoint.py @@ -1137,6 +1137,59 @@ def test_cease_active_reconciled_standby_atomically_opens_and_closes_gap( restarted.cease_active_reconciled_standby(**stale) +def test_cease_active_reconciled_standby_accepts_ambiguous_stop_ready_proof( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A dispatched STOP may receive its terminal READY only after restart.""" + + binding = _binding() + store = _store(tmp_path, monkeypatch) + _prepare(store, binding) + active = _activate(store, binding) + kwargs = _cease_active_reconciled_standby_kwargs( + active, + terminal_state="ready", + transition_id="transition-cease-active-ambiguous-stop-ready", + ) + status = kwargs["cessation_status_proof"] + prior_physical = active.current_active_physical_proof + assert prior_physical is not None + ambiguous_stop = ActiveAcquisitionRecoveryPhysicalLineageProof( + ledger_schema_version=( + ACTIVE_ACQUISITION_RECOVERY_PHYSICAL_LEDGER_SCHEMA + ), + ledger_revision=prior_physical.ledger_revision + 1, + proof_id="physical-ambiguous-stop-ready", + operation_id="operation-stop-ambiguous-ready", + original_start_operation_id=START_OPERATION_ID, + parent_operation_id=START_OPERATION_ID, + acquisition_id=ACQUISITION_ID, + action="stop", + resolution="physical-standby-observed", + payload_sha256=STOP_PAYLOAD_SHA256, + original_start_payload_sha256=START_PAYLOAD_SHA256, + reconciliation_kind="ambiguous-outcome", + reconciliation_resolution="physical-standby-observed", + status_message_sha256=status.status_message_sha256, + observed_session_state="ready", + binding=status.binding, + composite_complete=False, + edge_terminal=True, + late_start_excluded=True, + stop_fence="none", + observed_at_utc=status.observed_at_utc, + reconciliation_original_attempt_sha256="8" * 64, + ) + kwargs["cessation_physical_proof"] = ambiguous_stop + + ceased = store.cease_active_reconciled_standby(**kwargs) + + assert ceased.state == "ceased" + assert ceased.cessation_status_proof == status + assert ceased.cessation_physical_proof == ambiguous_stop + + def test_cease_active_reconciled_standby_atomically_closes_exact_open_gap( tmp_path: Path, monkeypatch: pytest.MonkeyPatch,