fix(k1): close verified stop checkpoint gap

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

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"

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(

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,