fix(k1): settle reopened standby after reset

src/k1link/device_plugins/xgrids_k1/physical_command_ledger.py

@@ -2625,6 +2625,9 @@ class PhysicalCommandLedger:
                 and latest_recovery.resolution == "physical-standby-observed"
                 and latest_recovery.observation.session_state == "scan_over"
             )
+            reopened_physical_state_requires_reconciliation = (
+                current.reopened_physical_state_requires_reconciliation
+            )
             if current.reconciliations:
                 latest = current.reconciliations[-1]
                 if latest.reconciliation_id == reconciliation_id:
@@ -2642,6 +2645,7 @@ class PhysicalCommandLedger:
             if (
                 latest_recovery is not None
                 and latest_recovery.resolution == ("physical-standby-observed")
+                and not reopened_physical_state_requires_reconciliation
                 and not (
                     awaiting_ready_after_scan_over
                     and observation.session_state == "ready"
@@ -2654,7 +2658,7 @@ class PhysicalCommandLedger:
                 )
             if (
                 current.reconciled_physical_state != "active"
-                and not current.reopened_physical_state_requires_reconciliation
+                and not reopened_physical_state_requires_reconciliation
                 and not awaiting_ready_after_scan_over
             ):
                 raise PhysicalCommandTransitionError(
@@ -3808,6 +3812,41 @@ def _latest_current_classified_stop_reopen(
     return matches[-1] if matches else None
 
 
+def _is_reopened_classified_stop_standby_settlement(
+    record: PhysicalCommandRecord,
+    reconciliation: PhysicalCommandReconciliation,
+    *,
+    settlement_index: int,
+) -> bool:
+    """Accept one fresh READY that settles a reopened READY-classified STOP."""
+
+    classification = _prepared_stop_classification(record)
+    if not (
+        classification is not None
+        and classification.resolution == "physical-standby-observed"
+        and reconciliation.kind == "resolved-active-cessation"
+        and reconciliation.resolution == "physical-standby-observed"
+        and reconciliation.original_attempt.operation_id == record.operation_id
+        and reconciliation.observation.session_state == "ready"
+        and not reconciliation.observation.project_bound
+        and not reconciliation.observation.init_ready
+    ):
+        return False
+    matching_reopens = [
+        (reopening, retirement)
+        for reopening in record.operator_reconciliation_reopens
+        for retirement in record.operator_retirements
+        if reopening.retirement_id == retirement.retirement_id
+        and retirement.original_attempt.operation_id == record.operation_id
+        and retirement.original_attempt.resolution == "not-dispatched"
+    ]
+    if settlement_index >= len(matching_reopens):
+        return False
+    reopening, _ = matching_reopens[settlement_index]
+    reconciled_at = _as_datetime(reconciliation.reconciled_at_utc)
+    return reconciled_at >= _as_datetime(reopening.reopened_at_utc)
+
+
 def _reopened_resolved_start_attempt(
     record: PhysicalCommandRecord,
 ) -> PhysicalCommandAttemptAudit | None:
@@ -4929,6 +4968,7 @@ def _validate_record_semantics(record: PhysicalCommandRecord) -> None:
             raise ValueError("physical reconciliation left the identity/profile chain")
     active_projects_by_operation: dict[str, str | None] = {}
     scan_over_ready_pending_by_operation: dict[str, bool] = {}
+    reopened_standby_settlements_by_operation: dict[str, int] = {}
     for reconciliation in record.reconciliations:
         operation_id = reconciliation.original_attempt.operation_id
         if (
@@ -4959,13 +4999,29 @@ def _validate_record_semantics(record: PhysicalCommandRecord) -> None:
                 raise ValueError("resolved-active rebind changed or lacked the active project")
             scan_over_ready_pending_by_operation[operation_id] = False
         elif reconciliation.kind == "resolved-active-cessation":
-            if active_projects_by_operation.get(operation_id) is None and not (
+            standby_after_scan_over = bool(
                 scan_over_ready_pending_by_operation.get(operation_id) is True
                 and reconciliation.observation.session_state == "ready"
                 and not reconciliation.observation.project_bound
                 and not reconciliation.observation.init_ready
+            )
+            if (
+                active_projects_by_operation.get(operation_id) is None
+                and not standby_after_scan_over
             ):
-                raise ValueError("resolved-active cessation lacks a prior active state")
+                settlement_index = reopened_standby_settlements_by_operation.get(
+                    operation_id,
+                    0,
+                )
+                if not _is_reopened_classified_stop_standby_settlement(
+                    record,
+                    reconciliation,
+                    settlement_index=settlement_index,
+                ):
+                    raise ValueError("resolved-active cessation lacks a prior active state")
+                reopened_standby_settlements_by_operation[operation_id] = (
+                    settlement_index + 1
+                )
             active_projects_by_operation[operation_id] = None
             scan_over_ready_pending_by_operation[operation_id] = (
                 reconciliation.observation.session_state == "scan_over"

tests/test_xgrids_physical_command_ledger.py

@@ -2084,6 +2084,65 @@ def test_ready_prepared_stop_classification_is_retireable_only_by_scenario_reset
     )
 
 
+def test_ready_prepared_stop_reset_reopen_accepts_fresh_ready(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A post-reset reopen must consume a new READY, not reject the old audit."""
+
+    ledger = _ledger(tmp_path, monkeypatch)
+    _prepare_start(ledger)
+    _complete_start(ledger)
+    _prepare_stop(ledger)
+    _, classified = _prepared_stop_classification(ledger, "ready")
+    assert isinstance(classified, ledger_module.PhysicalCommandRecord)
+
+    retired = ledger.retire_unavailable_target(
+        retirement_id="prepared-stop-ready-reset-retirement",
+        expected_operation_id=classified.operation_id,
+        expected_revision=classified.revision,
+        expected_transport_ref=classified.connection.transport_ref,
+        reason="connection-scenario-reset-by-operator",
+    )
+    reopened = ledger.reopen_retired_reconciliation(
+        reopening_id="prepared-stop-ready-reset-reopen",
+        expected_revision=retired.revision,
+        expected_retirement_id="prepared-stop-ready-reset-retirement",
+        expected_transport_ref=retired.connection.transport_ref,
+        expected_discovery_generation=120,
+        reason="reset-network-intent-read-only-settlement",
+    )
+    assert reopened.reopened_physical_state_requires_reconciliation is True
+
+    fresh_connection = _connection(
+        control_session_id="prepared-stop-ready-reset-fresh-control",
+        host_path_epoch=9,
+        producer_generation=13,
+    )
+    settled = ledger.observe_resolved_active_standby(
+        STOP_OPERATION,
+        reconciliation_id="prepared-stop-ready-reset-fresh-ready",
+        verified_binding=_verified_binding(
+            connection=fresh_connection,
+            verification_id="prepared-stop-ready-reset-fresh-ready.device-info",
+        ),
+        observation=_status(
+            "ready",
+            connection=fresh_connection,
+            source="explicit-read-only-reconciliation",
+            observed_at="2026-08-14T11:11:00.000Z",
+        ),
+    )
+
+    assert settled.resolution == "not-dispatched"
+    assert settled.original_command_outcome == "not-dispatched"
+    assert settled.reconciled_physical_state == "standby"
+    assert settled.reopened_physical_state_requires_reconciliation is False
+    assert PhysicalCommandLedger(tmp_path / "repository", clock=_clock).snapshot().record == (
+        settled
+    )
+
+
 def test_scenario_reset_retires_effective_active_prepared_stop_classification(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,