Enable safe NPM replica convergence

dns_sync/executor.py

@@ -314,6 +314,17 @@ def _required_capability(
         )
 
 
+def _required_retirement_support(
+    adapter: ExecutionAdapter,
+) -> None:
+    retire = getattr(adapter, "retire", None)
+    if not callable(retire):
+        raise RuntimeError(
+            f"{adapter.endpoint.identity} adapter does not support "
+            "deferred retirement"
+        )
+
+
 def _preflight(
     adapters: Iterable[ExecutionAdapter],
     plans: Iterable[EndpointPlan],
@@ -345,6 +356,9 @@ def _preflight(
                 f"No adapter for endpoint plan {plan.endpoint_id}"
             )
 
+        if plan.retirements:
+            _required_retirement_support(adapter)
+
         for retirement in plan.retirements:
             planned_retirements.append(
                 (
@@ -381,6 +395,9 @@ def _preflight(
                     "Executable plan action requires desired state"
                 )
 
+            if action.operation == PlanOperation.REPLACE:
+                _required_retirement_support(adapter)
+
             object_type = action.desired.object_type
             _required_capability(
                 adapter,

dns_sync/orchestrator.py

@@ -21,6 +21,7 @@ from .persistence import SQLiteStateStore
 from .projections import project_proxy_hosts_to_dns
 from .reconciliation import (
     build_endpoint_plans,
+    merge_desired_with_previous,
     reconcile_observations,
 )
 
@@ -315,6 +316,17 @@ def discover_reconcile_plan(
         else []
     )
 
+    if certificate_authoritative:
+        previous_certificates = [
+            row.object
+            for row in store.load_desired_state(CERTIFICATES)
+        ]
+
+        npm_desired = merge_desired_with_previous(
+            npm_desired,
+            previous_certificates,
+        )
+
     proxy_observations = _authoritative_observations(
         observations,
         live_pairs,

dns_sync/plugins/adapters/npm.py

@@ -43,6 +43,10 @@ class Plugin:
                 PROXY_HOSTS,
                 CapabilityOperation.DISCOVER,
             ),
+            Capability(
+                PROXY_HOSTS,
+                CapabilityOperation.APPLY,
+            ),
             Capability(
                 PROXY_HOSTS,
                 CapabilityOperation.VERIFY,
@@ -51,6 +55,10 @@ class Plugin:
                 CERTIFICATES,
                 CapabilityOperation.DISCOVER,
             ),
+            Capability(
+                CERTIFICATES,
+                CapabilityOperation.APPLY,
+            ),
             Capability(
                 CERTIFICATES,
                 CapabilityOperation.VERIFY,

dns_sync/reconciliation.py

@@ -54,6 +54,46 @@ class ReconciliationPolicy(Protocol):
     ) -> bool: ...
 
 
+def merge_desired_with_previous(
+    fresh: Iterable[NormalizedObject],
+    previous: Iterable[NormalizedObject],
+    policies: Mapping[str, ReconciliationPolicy] | None = None,
+) -> list[NormalizedObject]:
+    selected_policies = (
+        DEFAULT_POLICIES
+        if policies is None
+        else policies
+    )
+
+    previous_by_identity = {
+        (object.object_type, object.logical_key): object
+        for object in previous
+    }
+
+    merged: list[NormalizedObject] = []
+
+    for candidate in fresh:
+        identity = (
+            candidate.object_type,
+            candidate.logical_key,
+        )
+        prior = previous_by_identity.get(identity)
+        policy = selected_policies.get(candidate.object_type)
+
+        merger = (
+            getattr(policy, "merge_desired", None)
+            if policy is not None
+            else None
+        )
+
+        if prior is not None and callable(merger):
+            candidate = merger(prior, candidate)
+
+        merged.append(candidate)
+
+    return merged
+
+
 def _observation_sort_key(
     observation: Observation,
 ) -> tuple[str, str, str]:
@@ -343,6 +383,26 @@ class CertificateNewestPolicy:
             == self._portable_state(desired)
         )
 
+    def merge_desired(
+        self,
+        previous: NormalizedObject,
+        fresh: NormalizedObject,
+    ) -> NormalizedObject:
+        if (
+            previous.object_type != CERTIFICATES
+            or fresh.object_type != CERTIFICATES
+            or previous.logical_key != fresh.logical_key
+        ):
+            raise ReconciliationConflict(
+                "Certificate desired-state merge requires "
+                "matching certificate identities"
+            )
+
+        if self._newest_key(previous) > self._newest_key(fresh):
+            return previous
+
+        return fresh
+
     @staticmethod
     def _equivalent_selection_key(
         object: NormalizedObject,

tests/test_executor.py

@@ -655,6 +655,148 @@ class ExecutionTests(unittest.TestCase):
                 )
             self.assertEqual(store.load_applied_state(), [])
 
+    def test_preflight_requires_retire_for_planned_retirement_before_mutation(
+        self,
+    ):
+        endpoint = Endpoint("npm", "192.168.1.4:81")
+        inner = FakeAdapter(
+            endpoint,
+            capabilities(CERTIFICATES),
+        )
+
+        class NoRetireAdapter:
+            def __init__(self, wrapped):
+                self.endpoint = wrapped.endpoint
+                self.capabilities = wrapped.capabilities
+                self.wrapped = wrapped
+
+            def apply(self, plan_action):
+                return self.wrapped.apply(plan_action)
+
+            def verify(self, object):
+                return self.wrapped.verify(object)
+
+        adapter = NoRetireAdapter(inner)
+
+        old_certificate = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            local_key="67",
+            payload={
+                "domain_names": ["app.example.com"],
+            },
+        )
+
+        plan = EndpointPlan(
+            endpoint.identity,
+            (
+                action(
+                    endpoint.identity,
+                    PlanOperation.CREATE,
+                    object_type=CERTIFICATES,
+                    logical_key="other.example.com",
+                ),
+            ),
+            retirements=(
+                RetireRequest(
+                    object=old_certificate,
+                ),
+            ),
+        )
+
+        with SQLiteStateStore(":memory:") as store:
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "does not support deferred retirement",
+            ):
+                execute_endpoint_plans(
+                    [adapter],
+                    [plan],
+                    store,
+                    executed_at=NOW,
+                )
+
+            self.assertEqual(
+                store.load_applied_state(),
+                [],
+            )
+
+        self.assertEqual(inner.events, [])
+        self.assertEqual(inner.applied_actions, [])
+
+    def test_preflight_requires_retire_for_replace_before_mutation(
+        self,
+    ):
+        endpoint = Endpoint("npm", "192.168.1.4:81")
+        inner = FakeAdapter(
+            endpoint,
+            capabilities(CERTIFICATES),
+        )
+
+        class NoRetireAdapter:
+            def __init__(self, wrapped):
+                self.endpoint = wrapped.endpoint
+                self.capabilities = wrapped.capabilities
+                self.wrapped = wrapped
+
+            def apply(self, plan_action):
+                return self.wrapped.apply(plan_action)
+
+            def verify(self, object):
+                return self.wrapped.verify(object)
+
+        adapter = NoRetireAdapter(inner)
+
+        current = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            local_key="67",
+            payload={
+                "domain_names": ["app.example.com"],
+                "expires_on": "2026-11-01T00:00:00+00:00",
+            },
+        )
+        desired = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "expires_on": "2026-12-01T00:00:00+00:00",
+            },
+        )
+
+        plan = EndpointPlan(
+            endpoint.identity,
+            (
+                PlanAction(
+                    endpoint_id=endpoint.identity,
+                    operation=PlanOperation.REPLACE,
+                    desired=desired,
+                    current=current,
+                ),
+            ),
+        )
+
+        with SQLiteStateStore(":memory:") as store:
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "does not support deferred retirement",
+            ):
+                execute_endpoint_plans(
+                    [adapter],
+                    [plan],
+                    store,
+                    executed_at=NOW,
+                )
+
+            self.assertEqual(
+                store.load_applied_state(),
+                [],
+            )
+
+        self.assertEqual(inner.events, [])
+        self.assertEqual(inner.applied_actions, [])
+
     def test_preflight_requires_verify_before_any_mutation(self):
         endpoint = Endpoint("verizon", "192.168.1.1:443")
         adapter = FakeAdapter(

tests/test_npm_adapter.py

@@ -94,7 +94,7 @@ class NpmAdapterTests(unittest.TestCase):
         self.assertIs(adapters[0].client, first_client)
         self.assertIs(adapters[1].client, second_client)
 
-    def test_capabilities_cover_discovery_and_verify_only(self):
+    def test_capabilities_cover_discovery_apply_and_verify(self):
         adapter = self.adapter(Mock())
 
         self.assertEqual(
@@ -105,6 +105,10 @@ class NpmAdapterTests(unittest.TestCase):
                         PROXY_HOSTS,
                         CapabilityOperation.DISCOVER,
                     ),
+                    Capability(
+                        PROXY_HOSTS,
+                        CapabilityOperation.APPLY,
+                    ),
                     Capability(
                         PROXY_HOSTS,
                         CapabilityOperation.VERIFY,
@@ -113,6 +117,10 @@ class NpmAdapterTests(unittest.TestCase):
                         CERTIFICATES,
                         CapabilityOperation.DISCOVER,
                     ),
+                    Capability(
+                        CERTIFICATES,
+                        CapabilityOperation.APPLY,
+                    ),
                     Capability(
                         CERTIFICATES,
                         CapabilityOperation.VERIFY,
@@ -121,21 +129,6 @@ class NpmAdapterTests(unittest.TestCase):
             ),
         )
 
-        self.assertNotIn(
-            Capability(
-                PROXY_HOSTS,
-                CapabilityOperation.APPLY,
-            ),
-            adapter.capabilities,
-        )
-        self.assertNotIn(
-            Capability(
-                CERTIFICATES,
-                CapabilityOperation.APPLY,
-            ),
-            adapter.capabilities,
-        )
-
     def test_discovers_proxy_hosts_with_endpoint_local_key(self):
         client = Mock()
         client.proxy_hosts.return_value = [

tests/test_orchestrator_authority.py

@@ -240,6 +240,109 @@ class ObservationAuthorityTests(unittest.TestCase):
             )
         )
 
+    def test_persisted_newer_certificate_cannot_regress_to_older_live_certificate(
+        self,
+    ):
+        endpoint = Endpoint(
+            "npm",
+            "192.168.1.3:81",
+        )
+
+        live = FakeAdapter(
+            endpoint=endpoint,
+            capabilities=capability_set(
+                CERTIFICATES,
+                apply=True,
+            ),
+            discovered={
+                CERTIFICATES: [
+                    discovered_certificate(
+                        "20",
+                        valid_from=(
+                            "2026-09-15T00:00:00+00:00"
+                        ),
+                        expires_on=(
+                            "2026-12-15T00:00:00+00:00"
+                        ),
+                    )
+                ]
+            },
+        )
+
+        persisted = normalized_certificate(
+            "99",
+            valid_from=(
+                "2026-09-19T00:00:00+00:00"
+            ),
+            expires_on=(
+                "2026-12-19T00:00:00+00:00"
+            ),
+        )
+
+        persisted = NormalizedObject(
+            object_type=persisted.object_type,
+            logical_key=persisted.logical_key,
+            payload=dict(persisted.payload),
+        )
+
+        with SQLiteStateStore(":memory:") as store:
+            store.replace_desired_state(
+                CERTIFICATES,
+                [persisted],
+                T1,
+            )
+
+            result = discover_reconcile_plan(
+                [live],
+                store,
+                target_ip="192.168.1.101",
+                run_at=T2,
+            )
+
+            stored = store.load_desired_state(
+                CERTIFICATES
+            )
+
+        certificates = [
+            object
+            for object in result.desired
+            if object.object_type == CERTIFICATES
+        ]
+
+        self.assertEqual(len(certificates), 1)
+        self.assertEqual(
+            certificates[0].payload["valid_from"],
+            "2026-09-19T00:00:00+00:00",
+        )
+        self.assertEqual(
+            certificates[0].payload["expires_on"],
+            "2026-12-19T00:00:00+00:00",
+        )
+
+        self.assertEqual(len(stored), 1)
+        self.assertEqual(
+            stored[0].object,
+            persisted,
+        )
+
+        self.assertEqual(len(result.plans), 1)
+        self.assertEqual(
+            result.plans[0].endpoint_id,
+            endpoint.identity,
+        )
+        self.assertEqual(
+            len(result.plans[0].actions),
+            1,
+        )
+        self.assertEqual(
+            result.plans[0].actions[0].operation.value,
+            "replace",
+        )
+        self.assertEqual(
+            result.plans[0].actions[0].desired,
+            persisted,
+        )
+
     def test_all_failed_sources_preserve_persisted_desired_without_planning(self):
         npm_endpoint = Endpoint(
             "npm",