Account for deferred retirements in apply budget

dns_sync/adapters.py

@@ -19,6 +19,7 @@ class CapabilityOperation(StrEnum):
 class PlanOperation(StrEnum):
     CREATE = "create"
     UPDATE = "update"
+    REPLACE = "replace"
     DELETE = "delete"
     NOOP = "noop"
     SKIP = "skip"
@@ -157,16 +158,69 @@ class NormalizedObject:
         )
 
 
+@dataclass(frozen=True)
+class RetireRequest:
+    object: NormalizedObject
+
+    def __post_init__(self) -> None:
+        if self.object.local_key is None:
+            raise ValueError(
+                "Retired object requires an endpoint-local key"
+            )
+
+
 @dataclass(frozen=True)
 class ApplyResult:
     object: NormalizedObject
     detail: str | None = None
+    retirements: tuple[RetireRequest, ...] = ()
 
     def __post_init__(self) -> None:
         if self.object.local_key is None:
             raise ValueError(
                 "Applied object requires an endpoint-local key"
             )
+
+        retirements = tuple(self.retirements)
+
+        if any(
+            not isinstance(retirement, RetireRequest)
+            for retirement in retirements
+        ):
+            raise ValueError(
+                "ApplyResult retirements must be RetireRequest instances"
+            )
+
+        seen: set[tuple[str, str]] = set()
+
+        for retirement in retirements:
+            retired = retirement.object
+            identity = (
+                retired.object_type,
+                retired.local_key,
+            )
+
+            if identity in seen:
+                raise ValueError(
+                    "ApplyResult retirements cannot contain duplicates"
+                )
+
+            seen.add(identity)
+
+            if (
+                retired.object_type == self.object.object_type
+                and retired.local_key == self.object.local_key
+            ):
+                raise ValueError(
+                    "ApplyResult cannot retire the newly applied object"
+                )
+
+        object.__setattr__(
+            self,
+            "retirements",
+            retirements,
+        )
+
         if self.detail is not None:
             detail = self.detail.strip()
             object.__setattr__(
@@ -198,3 +252,35 @@ class PlanAction:
 class EndpointPlan:
     endpoint_id: str
     actions: tuple[PlanAction, ...]
+    retirements: tuple[RetireRequest, ...] = ()
+
+    def __post_init__(self) -> None:
+        retirements = tuple(self.retirements)
+
+        if any(
+            not isinstance(retirement, RetireRequest)
+            for retirement in retirements
+        ):
+            raise ValueError(
+                "EndpointPlan retirements must be "
+                "RetireRequest instances"
+            )
+
+        identities = [
+            (
+                retirement.object.object_type,
+                retirement.object.local_key,
+            )
+            for retirement in retirements
+        ]
+
+        if len(set(identities)) != len(identities):
+            raise ValueError(
+                "EndpointPlan retirements cannot contain duplicates"
+            )
+
+        object.__setattr__(
+            self,
+            "retirements",
+            retirements,
+        )

dns_sync/cli.py

@@ -65,7 +65,7 @@ def build_parser():
         "--max-changes",
         type=_nonnegative_int,
         default=None,
-        help="Maximum CREATE/UPDATE actions allowed with --apply",
+        help="Maximum planned mutations allowed with --apply",
     )
 
     sub.add_parser("plugins", help="Show the configured source/provider plugins")
@@ -202,11 +202,21 @@ def run_reconcile(args):
         for action in plan.actions
     )
 
+    retirement_count = sum(
+        len(plan.retirements)
+        for plan in result.plans
+    )
+    mutation_count = (
+        change_count
+        + retirement_count
+    )
+
     print()
     print(
         "Plan summary: "
         f"{create_count} create, "
         f"{update_count} update, "
+        f"{retirement_count} retirement, "
         f"{noop_count} unchanged"
     )
 
@@ -221,10 +231,10 @@ def run_reconcile(args):
 
     if (
         args.max_changes is not None
-        and change_count > args.max_changes
+        and mutation_count > args.max_changes
     ):
         raise RuntimeError(
-            f"Refusing to apply {change_count} change(s); "
+            f"Refusing to apply {mutation_count} mutation(s); "
             f"--max-changes allows {args.max_changes}"
         )
 
@@ -241,6 +251,7 @@ def run_reconcile(args):
     print(
         "APPLY COMPLETE: "
         f"{len(execution.applied)} action(s) applied and verified; "
+        f"{len(execution.retired)} retirement(s) completed; "
         f"{len(execution.skipped)} unchanged/skipped."
     )
     return 0

dns_sync/executor.py

@@ -13,6 +13,7 @@ from .adapters import (
     ObjectReference,
     PlanAction,
     PlanOperation,
+    RetireRequest,
 )
 from .dependencies import order_plan_actions
 from .persistence import SQLiteStateStore
@@ -26,6 +27,8 @@ class ExecutionAdapter(Protocol):
 
     def verify(self, object: NormalizedObject) -> bool: ...
 
+    def retire(self, request: RetireRequest) -> None: ...
+
 
 @dataclass(frozen=True)
 class ExecutedAction:
@@ -39,6 +42,7 @@ class ExecutedAction:
 class ExecutionResult:
     applied: tuple[ExecutedAction, ...]
     skipped: tuple[PlanAction, ...]
+    retired: tuple[tuple[str, RetireRequest], ...]
 
 
 @dataclass
@@ -271,6 +275,30 @@ def _validated_apply_result(
             "desired references"
         )
 
+    if (
+        action.operation != PlanOperation.REPLACE
+        and result.retirements
+    ):
+        raise RuntimeError(
+            "Only REPLACE actions may request deferred retirement"
+        )
+
+    for retirement in result.retirements:
+        retired = retirement.object
+
+        if retired.local_key is None:
+            raise RuntimeError(
+                "Deferred retirement requires endpoint-local identity"
+            )
+
+        if (
+            retired.object_type == applied.object_type
+            and retired.local_key == applied.local_key
+        ):
+            raise RuntimeError(
+                "Deferred retirement cannot target applied object"
+            )
+
     return result
 
 
@@ -287,6 +315,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],
@@ -294,6 +333,7 @@ def _preflight(
     dict[str, ExecutionAdapter],
     list[PlanAction],
     list[PlanAction],
+    list[tuple[str, RetireRequest]],
 ]:
     adapters_by_endpoint: dict[str, ExecutionAdapter] = {}
     for adapter in adapters:
@@ -306,6 +346,9 @@ def _preflight(
 
     executable: list[PlanAction] = []
     skipped: list[PlanAction] = []
+    planned_retirements: list[
+        tuple[str, RetireRequest]
+    ] = []
 
     for plan in plans:
         adapter = adapters_by_endpoint.get(plan.endpoint_id)
@@ -314,6 +357,17 @@ 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(
+                (
+                    plan.endpoint_id,
+                    retirement,
+                )
+            )
+
         for action in plan.actions:
             if action.endpoint_id != plan.endpoint_id:
                 raise RuntimeError(
@@ -330,6 +384,7 @@ def _preflight(
             if action.operation not in {
                 PlanOperation.CREATE,
                 PlanOperation.UPDATE,
+                PlanOperation.REPLACE,
             }:
                 raise RuntimeError(
                     "Unsupported executable plan operation: "
@@ -341,6 +396,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,
@@ -354,7 +412,12 @@ def _preflight(
             )
             executable.append(action)
 
-    return adapters_by_endpoint, executable, skipped
+    return (
+        adapters_by_endpoint,
+        executable,
+        skipped,
+        planned_retirements,
+    )
 
 
 def execute_endpoint_plans(
@@ -371,11 +434,21 @@ def execute_endpoint_plans(
         )
     timestamp = timestamp.astimezone(timezone.utc)
 
-    adapters_by_endpoint, executable, skipped = _preflight(
+    (
+        adapters_by_endpoint,
+        executable,
+        skipped,
+        planned_retirements,
+    ) = _preflight(
         adapters,
         plans,
     )
+
     applied: list[ExecutedAction] = []
+    pending_retirements: list[
+        tuple[str, RetireRequest]
+    ] = list(planned_retirements)
+
     local_references: dict[
         str,
         _LocalReferenceIndex,
@@ -389,6 +462,18 @@ def execute_endpoint_plans(
             )
         )
 
+    for action in skipped:
+        if (
+            action.operation == PlanOperation.NOOP
+            and action.current is not None
+            and action.current.local_key is not None
+        ):
+            _remember_local_object(
+                local_references[action.endpoint_id],
+                action.current,
+                authoritative=True,
+            )
+
     for action in order_plan_actions(executable):
         adapter = adapters_by_endpoint[action.endpoint_id]
         resolved_action = _resolve_action_references(
@@ -451,6 +536,14 @@ def execute_endpoint_plans(
             authoritative=True,
         )
 
+        for retirement in result.retirements:
+            pending_retirements.append(
+                (
+                    action.endpoint_id,
+                    retirement,
+                )
+            )
+
         applied.append(
             ExecutedAction(
                 endpoint_id=action.endpoint_id,
@@ -460,7 +553,46 @@ def execute_endpoint_plans(
             )
         )
 
+    retired_identities: set[
+        tuple[str, str, str]
+    ] = set()
+    retired_requests: list[
+        tuple[str, RetireRequest]
+    ] = []
+
+    for endpoint_id, retirement in pending_retirements:
+        retired = retirement.object
+        identity = (
+            endpoint_id,
+            retired.object_type,
+            retired.local_key,
+        )
+
+        if identity in retired_identities:
+            continue
+
+        retired_identities.add(identity)
+
+        adapter = adapters_by_endpoint[endpoint_id]
+
+        retire = getattr(adapter, "retire", None)
+
+        if retire is None:
+            raise RuntimeError(
+                f"{endpoint_id} adapter does not support "
+                "deferred retirement"
+            )
+
+        retire(retirement)
+        retired_requests.append(
+            (
+                endpoint_id,
+                retirement,
+            )
+        )
+
     return ExecutionResult(
         applied=tuple(applied),
         skipped=tuple(skipped),
+        retired=tuple(retired_requests),
     )

dns_sync/npm_client.py

@@ -223,6 +223,21 @@ class NpmClient:
             "proxy-host update",
         )
 
+    def delete_proxy_host(
+        self,
+        local_id: str | int,
+    ) -> None:
+        proxy_host_id = self._local_id(
+            local_id,
+            "proxy host",
+        )
+
+        self.request(
+            "DELETE",
+            f"/nginx/proxy-hosts/{proxy_host_id}",
+            expected_status=200,
+        )
+
     def certificate(
         self,
         local_id: str | int,
@@ -372,6 +387,21 @@ class NpmClient:
             f"Unsupported NPM certificate provider: {provider!r}"
         )
 
+    def delete_certificate(
+        self,
+        local_id: str | int,
+    ) -> None:
+        certificate_id = self._local_id(
+            local_id,
+            "certificate",
+        )
+
+        self.request(
+            "DELETE",
+            f"/nginx/certificates/{certificate_id}",
+            expected_status=200,
+        )
+
     def create_certificate(
         self,
         payload: dict[str, Any],

dns_sync/orchestrator.py

@@ -1,9 +1,10 @@
 from collections.abc import Iterable
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from datetime import datetime, timezone
 from typing import Protocol
 
 from .adapters import (
+    CERTIFICATES,
     DNS_RECORDS,
     PROXY_HOSTS,
     Capability,
@@ -19,7 +20,9 @@ from .object_normalizer import normalize_observations
 from .persistence import SQLiteStateStore
 from .projections import project_proxy_hosts_to_dns
 from .reconciliation import (
+    ReconciliationConflict,
     build_endpoint_plans,
+    merge_desired_with_previous,
     reconcile_observations,
 )
 
@@ -51,6 +54,12 @@ class ReconciliationRun:
     failures: tuple[DiscoveryFailure, ...]
 
 
+@dataclass(frozen=True)
+class _PlanningAdapterView:
+    endpoint: Endpoint
+    capabilities: frozenset[Capability]
+
+
 def _discovery_types(
     adapter: ReconciliationAdapter,
 ) -> tuple[str, ...]:
@@ -74,6 +83,216 @@ def _configured_observations(
     ]
 
 
+def _live_discovery_pairs(
+    adapters: Iterable[ReconciliationAdapter],
+    store: SQLiteStateStore,
+    attempted_at: datetime,
+) -> frozenset[tuple[str, str]]:
+    live: set[tuple[str, str]] = set()
+
+    for adapter in adapters:
+        endpoint_id = adapter.endpoint.identity
+
+        for object_type in _discovery_types(adapter):
+            status = store.observation_status(
+                endpoint_id,
+                object_type,
+            )
+
+            if (
+                status is not None
+                and status.is_live
+                and status.last_attempt_at == attempted_at
+            ):
+                live.add(
+                    (
+                        endpoint_id,
+                        object_type,
+                    )
+                )
+
+    return frozenset(live)
+
+
+def _authoritative_observations(
+    observations: Iterable[Observation],
+    live_pairs: frozenset[tuple[str, str]],
+    object_types: frozenset[str],
+) -> list[Observation]:
+    return [
+        observation
+        for observation in observations
+        if (
+            observation.object.object_type
+            in object_types
+            and (
+                observation.endpoint_id,
+                observation.object.object_type,
+            )
+            in live_pairs
+        )
+    ]
+
+
+def _certificate_expired(
+    object: NormalizedObject,
+    now: datetime,
+) -> bool:
+    value = object.payload.get("expires_on")
+
+    if value is None:
+        return False
+
+    if not isinstance(value, str):
+        raise ReconciliationConflict(
+            "Certificate expires_on must be normalized text"
+        )
+
+    normalized = value.strip()
+
+    if normalized.endswith("Z"):
+        normalized = normalized[:-1] + "+00:00"
+
+    try:
+        expires_at = datetime.fromisoformat(normalized)
+    except ValueError as exc:
+        raise ReconciliationConflict(
+            "Certificate expires_on is not a valid timestamp"
+        ) from exc
+
+    if expires_at.tzinfo is None:
+        expires_at = expires_at.replace(tzinfo=timezone.utc)
+
+    return expires_at.astimezone(timezone.utc) <= now
+
+
+def _filter_expired_certificate_dependencies(
+    observations: Iterable[Observation],
+    now: datetime,
+) -> list[Observation]:
+    rows = list(observations)
+
+    expired_local: set[tuple[str, str]] = set()
+
+    for observation in rows:
+        object = observation.object
+
+        if (
+            object.object_type != CERTIFICATES
+            or not _certificate_expired(object, now)
+        ):
+            continue
+
+        if object.local_key is None:
+            raise ReconciliationConflict(
+                "Expired observed certificate requires "
+                "endpoint-local identity"
+            )
+
+        expired_local.add(
+            (
+                observation.endpoint_id,
+                object.local_key,
+            )
+        )
+
+    filtered: list[Observation] = []
+
+    for observation in rows:
+        object = observation.object
+
+        if (
+            object.object_type == CERTIFICATES
+            and object.local_key is not None
+            and (
+                observation.endpoint_id,
+                object.local_key,
+            )
+            in expired_local
+        ):
+            continue
+
+        if object.object_type != PROXY_HOSTS:
+            filtered.append(observation)
+            continue
+
+        expired_references = tuple(
+            reference
+            for reference in object.references
+            if (
+                reference.object_type == CERTIFICATES
+                and reference.local_key is not None
+                and (
+                    observation.endpoint_id,
+                    reference.local_key,
+                )
+                in expired_local
+            )
+        )
+
+        if not expired_references:
+            filtered.append(observation)
+            continue
+
+        if object.payload.get("enabled") is True:
+            raise ReconciliationConflict(
+                "Enabled proxy host references an expired "
+                f"certificate: {object.logical_key!r} "
+                f"on {observation.endpoint_id}"
+            )
+
+        retained_references = tuple(
+            reference
+            for reference in object.references
+            if reference not in expired_references
+        )
+
+        filtered.append(
+            replace(
+                observation,
+                object=replace(
+                    object,
+                    references=retained_references,
+                ),
+            )
+        )
+
+    return filtered
+
+
+def _planning_adapter_views(
+    adapters: Iterable[ReconciliationAdapter],
+    live_pairs: frozenset[tuple[str, str]],
+) -> list[_PlanningAdapterView]:
+    views: list[_PlanningAdapterView] = []
+
+    for adapter in adapters:
+        endpoint_id = adapter.endpoint.identity
+
+        capabilities = frozenset(
+            capability
+            for capability in adapter.capabilities
+            if (
+                capability.operation
+                != CapabilityOperation.APPLY
+                or (
+                    endpoint_id,
+                    capability.object_type,
+                )
+                in live_pairs
+            )
+        )
+
+        views.append(
+            _PlanningAdapterView(
+                endpoint=adapter.endpoint,
+                capabilities=capabilities,
+            )
+        )
+
+    return views
+
+
 def _filter_projected_dns(
     observations: Iterable[Observation],
     *,
@@ -188,30 +407,127 @@ def discover_reconcile_plan(
         store,
         set(adapters_by_endpoint),
     )
-    proxy_observations = [
+
-        observation
+    # Cached observations remain visible, but only a successful
-        for observation in observations
+    # discovery from this exact run may establish desired state or
-        if observation.object.object_type == PROXY_HOSTS
+    # make an endpoint eligible for mutation.
-    ]
+    live_pairs = _live_discovery_pairs(
-    projected_dns = project_proxy_hosts_to_dns(
+        adapter_list,
-        proxy_observations,
+        store,
-        target_ip,
+        timestamp,
-        include_disabled=include_disabled,
     )
+
+    certificate_authoritative = any(
+        object_type == CERTIFICATES
+        for _, object_type in live_pairs
+    )
+    proxy_authoritative = any(
+        object_type == PROXY_HOSTS
+        for _, object_type in live_pairs
+    )
+
+    npm_observations = _authoritative_observations(
+        observations,
+        live_pairs,
+        frozenset(
+            {
+                CERTIFICATES,
+                PROXY_HOSTS,
+            }
+        ),
+    )
+
+    npm_observations = _filter_expired_certificate_dependencies(
+        npm_observations,
+        timestamp,
+    )
+
+    npm_desired = (
+        reconcile_observations(npm_observations)
+        if npm_observations
+        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,
+        frozenset({PROXY_HOSTS}),
+    )
+
+    projected_dns = (
+        project_proxy_hosts_to_dns(
+            proxy_observations,
+            target_ip,
+            include_disabled=include_disabled,
+        )
+        if proxy_authoritative
+        else []
+    )
+
     projected_dns = _filter_projected_dns(
         projected_dns,
         domain_matching=domain_matching,
         domain=domain,
     )
 
-    desired = reconcile_observations(projected_dns)
+    dns_desired = (
-    store.replace_desired_state(
+        reconcile_observations(projected_dns)
-        DNS_RECORDS,
+        if projected_dns
-        desired,
+        else []
-        timestamp,
     )
+
+    # Keep the reconciliation domains separate until each has
+    # produced portable desired state. Router DNS observations are
+    # current target state, never a desired-state authority.
+    desired = [
+        *npm_desired,
+        *dns_desired,
+    ]
+
+    authority_by_type = {
+        CERTIFICATES: certificate_authoritative,
+        PROXY_HOSTS: proxy_authoritative,
+        # DNS desired state is projected from proxy-host authority.
+        DNS_RECORDS: proxy_authoritative,
+    }
+
+    for object_type in (
+        CERTIFICATES,
+        PROXY_HOSTS,
+        DNS_RECORDS,
+    ):
+        if not authority_by_type[object_type]:
+            # No successful source discovery this run: retain the
+            # previously selected desired snapshot for diagnostics,
+            # but do not use it for this run's plans.
+            continue
+
+        store.replace_desired_state(
+            object_type,
+            [
+                object
+                for object in desired
+                if object.object_type == object_type
+            ],
+            timestamp,
+        )
+
     plans = build_endpoint_plans(
-        adapter_list,
+        _planning_adapter_views(
+            adapter_list,
+            live_pairs,
+        ),
         desired,
         observations,
     )

dns_sync/plugins/adapters/npm.py

@@ -3,11 +3,15 @@ from typing import Any
 from ...adapters import (
     CERTIFICATES,
     PROXY_HOSTS,
+    ApplyResult,
     Capability,
     CapabilityOperation,
     DiscoveredObject,
     NormalizedObject,
     ObjectReference,
+    PlanAction,
+    PlanOperation,
+    RetireRequest,
 )
 from ...certificates.material import CertificateMaterial
 from ...npm_client import NpmClient
@@ -39,6 +43,10 @@ class Plugin:
                 PROXY_HOSTS,
                 CapabilityOperation.DISCOVER,
             ),
+            Capability(
+                PROXY_HOSTS,
+                CapabilityOperation.APPLY,
+            ),
             Capability(
                 PROXY_HOSTS,
                 CapabilityOperation.VERIFY,
@@ -47,6 +55,10 @@ class Plugin:
                 CERTIFICATES,
                 CapabilityOperation.DISCOVER,
             ),
+            Capability(
+                CERTIFICATES,
+                CapabilityOperation.APPLY,
+            ),
             Capability(
                 CERTIFICATES,
                 CapabilityOperation.VERIFY,
@@ -312,9 +324,484 @@ class Plugin:
 
         return baseline
 
-    def apply(self, action):
+    @staticmethod
+    def _proxy_host_matches(
+        actual: NormalizedObject,
+        desired: NormalizedObject,
+        *,
+        local_key: str | None = None,
+    ) -> bool:
+        if actual.object_type != PROXY_HOSTS:
+            return False
+
+        if desired.object_type != PROXY_HOSTS:
+            return False
+
+        if local_key is not None and actual.local_key != local_key:
+            return False
+
+        if actual.logical_key != desired.logical_key:
+            return False
+
+        if actual.payload != desired.payload:
+            return False
+
+        if actual.references != desired.references:
+            return False
+
+        return True
+
+    @staticmethod
+    def _proxy_host_certificate_id(
+        desired: NormalizedObject,
+    ) -> int:
+        certificate_references = []
+
+        for reference in desired.references:
+            if (
+                reference.relation != "certificate"
+                or reference.object_type != CERTIFICATES
+            ):
+                raise RuntimeError(
+                    "NPM proxy host has unsupported object reference"
+                )
+
+            certificate_references.append(reference)
+
+        if not certificate_references:
+            return 0
+
+        if len(certificate_references) != 1:
+            raise RuntimeError(
+                "NPM proxy host requires at most one "
+                "certificate reference"
+            )
+
+        reference = certificate_references[0]
+
+        if reference.local_key is None:
+            raise RuntimeError(
+                "NPM proxy host certificate reference "
+                "was not resolved to an endpoint-local key"
+            )
+
+        try:
+            certificate_id = int(reference.local_key)
+        except (TypeError, ValueError) as exc:
+            raise RuntimeError(
+                "NPM proxy host certificate reference "
+                "has invalid endpoint-local key"
+            ) from exc
+
+        if certificate_id < 1:
+            raise RuntimeError(
+                "NPM proxy host certificate reference "
+                "has invalid endpoint-local key"
+            )
+
+        return certificate_id
+
+    def _proxy_host_api_payload(
+        self,
+        desired: NormalizedObject,
+    ) -> dict[str, Any]:
+        if desired.object_type != PROXY_HOSTS:
+            raise RuntimeError(
+                "Proxy-host API payload requires proxy-host state"
+            )
+
+        payload = desired.payload
+
+        domains = payload.get("domain_names")
+        if (
+            not isinstance(domains, list)
+            or not domains
+            or not all(
+                isinstance(value, str) and value
+                for value in domains
+            )
+        ):
+            raise RuntimeError(
+                "NPM proxy host requires domain_names"
+            )
+
+        forward_scheme = payload.get("forward_scheme")
+        if (
+            not isinstance(forward_scheme, str)
+            or not forward_scheme
+        ):
+            raise RuntimeError(
+                "NPM proxy host requires forward_scheme"
+            )
+
+        forward_host = payload.get("forward_host")
+        if (
+            not isinstance(forward_host, str)
+            or not forward_host
+        ):
+            raise RuntimeError(
+                "NPM proxy host requires forward_host"
+            )
+
+        forward_port = payload.get("forward_port")
+        if (
+            isinstance(forward_port, bool)
+            or not isinstance(forward_port, int)
+            or forward_port < 1
+            or forward_port > 65535
+        ):
+            raise RuntimeError(
+                "NPM proxy host requires valid forward_port"
+            )
+
+        api_payload: dict[str, Any] = {
+            "domain_names": list(domains),
+            "forward_scheme": forward_scheme,
+            "forward_host": forward_host,
+            "forward_port": forward_port,
+            "certificate_id": (
+                self._proxy_host_certificate_id(desired)
+            ),
+        }
+
+        for field in _PROXY_HOST_FIELDS:
+            if field in (
+                "domain_names",
+                "forward_scheme",
+                "forward_host",
+                "forward_port",
+                "certificate_id",
+            ):
+                continue
+
+            if field in payload:
+                api_payload[field] = payload[field]
+
+        return api_payload
+
+    def _create_proxy_host(
+        self,
+        desired: NormalizedObject,
+    ) -> ApplyResult:
+        payload = self._proxy_host_api_payload(
+            desired
+        )
+
+        created = self.client.create_proxy_host(
+            payload
+        )
+
+        local_key = self._local_key(
+            created,
+            "proxy host",
+        )
+
+        try:
+            row = self.client.proxy_host(local_key)
+
+            actual = normalize_discovered_object(
+                self._proxy_host_from_row(row)
+            )
+
+            if not self._proxy_host_matches(
+                actual,
+                desired,
+                local_key=local_key,
+            ):
+                raise RuntimeError(
+                    "Created NPM proxy host does not match "
+                    "desired state"
+                )
+
+        except Exception:
+            try:
+                self.client.delete_proxy_host(
+                    local_key
+                )
+            except Exception as rollback_error:
+                raise RuntimeError(
+                    "NPM proxy-host creation failed and "
+                    "rollback also failed"
+                ) from rollback_error
+
+            raise
+
+        return ApplyResult(
+            object=actual,
+            detail="created proxy host",
+        )
+
+    def _update_proxy_host(
+        self,
+        action: PlanAction,
+    ) -> ApplyResult:
+        desired = action.desired
+        current = action.current
+
+        if desired is None:
+            raise RuntimeError(
+                "NPM proxy-host update requires desired state"
+            )
+
+        if (
+            current is None
+            or current.object_type != PROXY_HOSTS
+            or current.local_key is None
+        ):
+            raise RuntimeError(
+                "NPM proxy-host update requires "
+                "current endpoint-local state"
+            )
+
+        if current.logical_key != desired.logical_key:
+            raise RuntimeError(
+                "NPM proxy-host update identity changed"
+            )
+
+        local_key = current.local_key
+        payload = self._proxy_host_api_payload(
+            desired
+        )
+
+        self.client.update_proxy_host(
+            local_key,
+            payload,
+        )
+
+        row = self.client.proxy_host(local_key)
+
+        actual = normalize_discovered_object(
+            self._proxy_host_from_row(row)
+        )
+
+        if not self._proxy_host_matches(
+            actual,
+            desired,
+            local_key=local_key,
+        ):
+            raise RuntimeError(
+                "Updated NPM proxy host does not match "
+                "desired state"
+            )
+
+        return ApplyResult(
+            object=actual,
+            detail="updated proxy host",
+        )
+
+    def apply(
+        self,
+        action: PlanAction,
+    ) -> ApplyResult | None:
+        desired = action.desired
+
+        if desired is None:
+            raise RuntimeError(
+                "NPM apply requires desired state"
+            )
+
+        if action.endpoint_id != self.endpoint.identity:
+            raise RuntimeError(
+                "NPM apply action endpoint does not match adapter"
+            )
+
+        if (
+            desired.object_type == CERTIFICATES
+            and action.operation == PlanOperation.CREATE
+        ):
+            return self._create_certificate(desired)
+
+        if (
+            desired.object_type == CERTIFICATES
+            and action.operation == PlanOperation.REPLACE
+        ):
+            return self._replace_certificate(action)
+
+        if (
+            desired.object_type == PROXY_HOSTS
+            and action.operation == PlanOperation.CREATE
+        ):
+            return self._create_proxy_host(desired)
+
+        if (
+            desired.object_type == PROXY_HOSTS
+            and action.operation == PlanOperation.UPDATE
+        ):
+            return self._update_proxy_host(action)
+
         raise RuntimeError(
-            f"{self.endpoint.identity} does not support apply yet"
+            f"{self.endpoint.identity} does not support "
+            f"{action.operation} for {desired.object_type}"
+        )
+
+    def _replace_certificate(
+        self,
+        action: PlanAction,
+    ) -> ApplyResult:
+        desired = action.desired
+        current = action.current
+
+        if desired is None or desired.object_type != CERTIFICATES:
+            raise RuntimeError(
+                "NPM certificate replace requires desired certificate"
+            )
+
+        if current is None or current.object_type != CERTIFICATES:
+            raise RuntimeError(
+                "NPM certificate replace requires current certificate"
+            )
+
+        if current.local_key is None:
+            raise RuntimeError(
+                "NPM certificate replace requires current local id"
+            )
+
+        if current.logical_key != desired.logical_key:
+            raise RuntimeError(
+                "NPM certificate replace identity mismatch"
+            )
+
+        result = self._create_certificate(desired)
+
+        return ApplyResult(
+            object=result.object,
+            detail="replaced certificate",
+            retirements=(
+                RetireRequest(
+                    object=current,
+                ),
+            ),
+        )
+
+    def retire(
+        self,
+        request: RetireRequest,
+    ) -> None:
+        retired = request.object
+
+        if retired.object_type != CERTIFICATES:
+            raise RuntimeError(
+                "NPM deferred retirement only supports certificates"
+            )
+
+        if retired.local_key is None:
+            raise RuntimeError(
+                "NPM certificate retirement requires local id"
+            )
+
+        self.client.delete_certificate(
+            retired.local_key
+        )
+
+    def _create_certificate(
+        self,
+        desired: NormalizedObject,
+    ) -> ApplyResult:
+        material = self._certificate_material_source(
+            desired
+        )
+
+        domains = desired.payload.get("domain_names")
+
+        if (
+            not isinstance(domains, list)
+            or not domains
+            or not all(
+                isinstance(value, str) and value
+                for value in domains
+            )
+        ):
+            raise RuntimeError(
+                "NPM certificate create requires domain_names"
+            )
+
+        nice_name = desired.payload.get("nice_name")
+
+        if not isinstance(nice_name, str):
+            nice_name = ""
+
+        nice_name = nice_name.strip()
+
+        if not nice_name:
+            nice_name = (
+                "dns-sync: "
+                + ", ".join(domains)
+            )
+
+        created = self.client.create_certificate(
+            {
+                # This node is receiving existing certificate
+                # material. It must not independently perform ACME
+                # issuance.
+                "provider": "other",
+                "domain_names": list(domains),
+                "nice_name": nice_name,
+                "meta": {
+                    "cluster_managed": True,
+                    "acme_renew": False,
+                    "dns_challenge": False,
+                },
+            }
+        )
+
+        created_id = created.get("id")
+
+        if (
+            isinstance(created_id, bool)
+            or not isinstance(created_id, int)
+            or created_id < 1
+        ):
+            raise RuntimeError(
+                "NPM certificate create returned invalid id"
+            )
+
+        local_key = str(created_id)
+
+        try:
+            self.client.upload_certificate(
+                local_key,
+                certificate=material.certificate,
+                certificate_key=material.certificate_key,
+                intermediate_certificate=(
+                    material.intermediate_certificate
+                ),
+            )
+
+            row = self.client.certificate(local_key)
+
+            actual = normalize_discovered_object(
+                self._certificate_from_row(row)
+            )
+
+            if actual.local_key != local_key:
+                raise RuntimeError(
+                    "Created NPM certificate returned "
+                    "unexpected local id"
+                )
+
+            if not self._portable_certificate_matches(
+                actual,
+                desired,
+            ):
+                raise RuntimeError(
+                    "Created NPM certificate does not match "
+                    "desired portable certificate state"
+                )
+
+        except Exception:
+            try:
+                self.client.delete_certificate(local_key)
+            except Exception as rollback_error:
+                raise RuntimeError(
+                    "NPM certificate creation failed and "
+                    "rollback also failed"
+                ) from rollback_error
+
+            raise
+
+        return ApplyResult(
+            object=actual,
+            detail="created certificate",
         )
 
     def verify(self, object: NormalizedObject) -> bool:
@@ -329,7 +816,11 @@ class Plugin:
                 self._proxy_host_from_row(row)
             )
 
-            return actual == object
+            return self._proxy_host_matches(
+                actual,
+                object,
+                local_key=object.local_key,
+            )
 
         if object.object_type == CERTIFICATES:
             row = self.client.certificate(

dns_sync/reconciliation.py

@@ -1,17 +1,22 @@
 from collections import defaultdict
 from collections.abc import Iterable, Mapping, Sequence
+from dataclasses import replace
 from typing import Protocol
 
 from .adapters import (
+    CERTIFICATES,
+    PROXY_HOSTS,
     Capability,
     CapabilityOperation,
     DNS_RECORDS,
     Endpoint,
     EndpointPlan,
     NormalizedObject,
+    ObjectReference,
     Observation,
     PlanAction,
     PlanOperation,
+    RetireRequest,
 )
 from .dependencies import order_plan_actions
 
@@ -34,6 +39,7 @@ class PlanningAdapter(Protocol):
 
 class ReconciliationPolicy(Protocol):
     object_type: str
+    mutation_operation: PlanOperation
 
     def reconcile(
         self,
@@ -41,6 +47,52 @@ class ReconciliationPolicy(Protocol):
         observations: Sequence[Observation],
     ) -> NormalizedObject: ...
 
+    def equivalent(
+        self,
+        current: NormalizedObject,
+        desired: NormalizedObject,
+    ) -> 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,
@@ -52,6 +104,94 @@ def _observation_sort_key(
     )
 
 
+def logicalize_observation_references(
+    observations: Iterable[Observation],
+) -> tuple[Observation, ...]:
+    rows = tuple(observations)
+
+    local_targets: dict[
+        tuple[str, str, str],
+        str,
+    ] = {}
+
+    for observation in rows:
+        object = observation.object
+
+        if object.local_key is None:
+            continue
+
+        identity = (
+            observation.endpoint_id,
+            object.object_type,
+            object.local_key,
+        )
+
+        existing = local_targets.get(identity)
+
+        if (
+            existing is not None
+            and existing != object.logical_key
+        ):
+            raise ReconciliationConflict(
+                "Conflicting endpoint-local object identity for "
+                f"{observation.endpoint_id} "
+                f"{object.object_type} "
+                f"local key {object.local_key!r}"
+            )
+
+        local_targets[identity] = object.logical_key
+
+    logicalized: list[Observation] = []
+
+    for observation in rows:
+        object = observation.object
+
+        references = []
+
+        for reference in object.references:
+            if reference.logical_key is not None:
+                references.append(reference)
+                continue
+
+            target = (
+                observation.endpoint_id,
+                reference.object_type,
+                reference.local_key,
+            )
+            logical_key = local_targets.get(target)
+
+            if logical_key is None:
+                raise ReconciliationConflict(
+                    "Unresolved endpoint-local reference from "
+                    f"{observation.endpoint_id} "
+                    f"{object.object_type} "
+                    f"{object.logical_key!r}: "
+                    f"{reference.relation} -> "
+                    f"{reference.object_type} "
+                    f"local key {reference.local_key!r}"
+                )
+
+            references.append(
+                ObjectReference(
+                    relation=reference.relation,
+                    object_type=reference.object_type,
+                    logical_key=logical_key,
+                )
+            )
+
+        logicalized.append(
+            replace(
+                observation,
+                object=replace(
+                    object,
+                    references=tuple(references),
+                ),
+            )
+        )
+
+    return tuple(logicalized)
+
+
 def group_observations(
     observations: Iterable[Observation],
 ) -> dict[ObservationGroupKey, tuple[Observation, ...]]:
@@ -83,6 +223,7 @@ def group_observations(
 
 class DnsUnionPolicy:
     object_type = DNS_RECORDS
+    mutation_operation = PlanOperation.UPDATE
 
     def reconcile(
         self,
@@ -116,9 +257,294 @@ class DnsUnionPolicy:
             payload=dict(first.payload),
         )
 
+    def equivalent(
+        self,
+        current: NormalizedObject,
+        desired: NormalizedObject,
+    ) -> bool:
+        return (
+            current.object_type == desired.object_type
+            and current.logical_key == desired.logical_key
+            and current.payload == desired.payload
+            and current.references == desired.references
+        )
+
+
+class CertificateNewestPolicy:
+    object_type = CERTIFICATES
+    mutation_operation = PlanOperation.REPLACE
+
+    _PORTABLE_FIELDS = (
+        "domain_names",
+        "valid_from",
+        "expires_on",
+    )
+
+    @classmethod
+    def _portable_state(
+        cls,
+        object: NormalizedObject,
+    ) -> tuple[object, ...]:
+        return tuple(
+            object.payload.get(field)
+            for field in cls._PORTABLE_FIELDS
+        )
+
+    @staticmethod
+    def _date_key(
+        value: object,
+        field: str,
+    ) -> tuple[int, str]:
+        if value is None:
+            return (0, "")
+
+        if not isinstance(value, str):
+            raise ReconciliationConflict(
+                f"Certificate {field} must be normalized text"
+            )
+
+        return (1, value)
+
+    @classmethod
+    def _newest_key(
+        cls,
+        object: NormalizedObject,
+    ) -> tuple[
+        tuple[int, str],
+        tuple[int, str],
+    ]:
+        return (
+            cls._date_key(
+                object.payload.get("valid_from"),
+                "valid_from",
+            ),
+            cls._date_key(
+                object.payload.get("expires_on"),
+                "expires_on",
+            ),
+        )
+
+    def reconcile(
+        self,
+        logical_key: str,
+        observations: Sequence[Observation],
+    ) -> NormalizedObject:
+        if not observations:
+            raise RuntimeError(
+                "Certificate policy requires at least one observation"
+            )
+
+        newest_key = max(
+            self._newest_key(observation.object)
+            for observation in observations
+        )
+
+        newest = [
+            observation
+            for observation in observations
+            if self._newest_key(observation.object) == newest_key
+        ]
+
+        baseline = self._portable_state(newest[0].object)
+
+        conflicting_endpoints = {
+            observation.endpoint_id
+            for observation in newest
+            if self._portable_state(observation.object) != baseline
+        }
+
+        if conflicting_endpoints:
+            conflicting_endpoints.add(newest[0].endpoint_id)
+
+            raise ReconciliationConflict(
+                f"Conflicting {CERTIFICATES} observations for "
+                f"{logical_key!r} at newest issuance from "
+                + ", ".join(sorted(conflicting_endpoints))
+            )
+
+        winner = newest[0].object
+
+        return NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key=logical_key,
+            payload=dict(winner.payload),
+        )
+
+    def equivalent(
+        self,
+        current: NormalizedObject,
+        desired: NormalizedObject,
+    ) -> bool:
+        return (
+            current.object_type == CERTIFICATES
+            and desired.object_type == CERTIFICATES
+            and current.logical_key == desired.logical_key
+            and self._portable_state(current)
+            == 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,
+    ) -> tuple[int, str]:
+        # Preserve a native ACME-renewable certificate over an
+        # imported replica when both contain the same portable
+        # certificate state.
+        provider = object.payload.get("provider")
+
+        return (
+            1 if provider == "letsencrypt" else 0,
+            object.local_key or "",
+        )
+
+    def select_current(
+        self,
+        observations: Sequence[Observation],
+        desired: NormalizedObject,
+    ) -> tuple[
+        NormalizedObject,
+        tuple[RetireRequest, ...],
+    ]:
+        if not observations:
+            raise PlanningConflict(
+                "Certificate current-state selection requires "
+                "at least one observation"
+            )
+
+        candidates = [
+            observation.object
+            for observation in observations
+        ]
+
+        if any(
+            candidate.local_key is None
+            for candidate in candidates
+        ):
+            raise PlanningConflict(
+                "Observed certificate requires endpoint-local identity"
+            )
+
+        matching = [
+            candidate
+            for candidate in candidates
+            if self.equivalent(
+                candidate,
+                desired,
+            )
+        ]
+
+        if matching:
+            selected = max(
+                matching,
+                key=self._equivalent_selection_key,
+            )
+        else:
+            # No local copy is already desired. Replace the newest
+            # local certificate and retire every other stale copy
+            # after the replacement and its dependents verify.
+            selected = max(
+                candidates,
+                key=lambda candidate: (
+                    self._newest_key(candidate),
+                    candidate.local_key or "",
+                ),
+            )
+
+        retirements = tuple(
+            RetireRequest(object=candidate)
+            for candidate in candidates
+            if candidate.local_key != selected.local_key
+        )
+
+        return selected, retirements
+
+
+class ProxyHostConsensusPolicy:
+    object_type = PROXY_HOSTS
+    mutation_operation = PlanOperation.UPDATE
+
+    def reconcile(
+        self,
+        logical_key: str,
+        observations: Sequence[Observation],
+    ) -> NormalizedObject:
+        if not observations:
+            raise RuntimeError(
+                "Proxy-host policy requires at least one observation"
+            )
+
+        first = observations[0].object
+
+        conflicting_endpoints = {
+            observation.endpoint_id
+            for observation in observations
+            if (
+                observation.object.payload != first.payload
+                or observation.object.references != first.references
+            )
+        }
+
+        if conflicting_endpoints:
+            conflicting_endpoints.add(observations[0].endpoint_id)
+
+            raise ReconciliationConflict(
+                f"Conflicting {PROXY_HOSTS} observations for "
+                f"{logical_key!r} from "
+                + ", ".join(sorted(conflicting_endpoints))
+            )
+
+        for reference in first.references:
+            if reference.local_key is not None:
+                raise ReconciliationConflict(
+                    "Proxy-host desired state cannot contain "
+                    "endpoint-local references"
+                )
+
+        return NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key=logical_key,
+            payload=dict(first.payload),
+            references=first.references,
+        )
+
+    def equivalent(
+        self,
+        current: NormalizedObject,
+        desired: NormalizedObject,
+    ) -> bool:
+        return (
+            current.object_type == PROXY_HOSTS
+            and desired.object_type == PROXY_HOSTS
+            and current.logical_key == desired.logical_key
+            and current.payload == desired.payload
+            and current.references == desired.references
+        )
+
 
 DEFAULT_POLICIES: Mapping[str, ReconciliationPolicy] = {
     DNS_RECORDS: DnsUnionPolicy(),
+    CERTIFICATES: CertificateNewestPolicy(),
+    PROXY_HOSTS: ProxyHostConsensusPolicy(),
 }
 
 
@@ -133,8 +559,12 @@ def reconcile_observations(
     )
     desired: list[NormalizedObject] = []
 
+    logicalized = logicalize_observation_references(
+        observations
+    )
+
     for (object_type, logical_key), rows in (
-        group_observations(observations).items()
+        group_observations(logicalized).items()
     ):
         policy = selected_policies.get(object_type)
         if policy is None:
@@ -156,6 +586,7 @@ def build_endpoint_plans(
     adapters: Iterable[PlanningAdapter],
     desired_objects: Iterable[NormalizedObject],
     observations: Iterable[Observation],
+    policies: Mapping[str, ReconciliationPolicy] | None = None,
 ) -> tuple[EndpointPlan, ...]:
     desired_by_identity: dict[
         tuple[str, str],
@@ -180,11 +611,40 @@ def build_endpoint_plans(
             )
         desired_by_identity[identity] = desired
 
+    selected_policies = (
+        DEFAULT_POLICIES
+        if policies is None
+        else policies
+    )
+
+    raw_observations = tuple(observations)
+
     current_by_identity: dict[
         tuple[str, str, str],
         list[Observation],
     ] = defaultdict(list)
-    for observation in observations:
+
+    raw_current_by_identity: dict[
+        tuple[str, str, str],
+        list[Observation],
+    ] = defaultdict(list)
+
+    for observation in raw_observations:
+        raw_current_by_identity[
+            (
+                observation.endpoint_id,
+                observation.object.object_type,
+                observation.object.logical_key,
+            )
+        ].append(observation)
+
+    logicalized_observations = (
+        logicalize_observation_references(
+            raw_observations
+        )
+    )
+
+    for observation in logicalized_observations:
         current_by_identity[
             (
                 observation.endpoint_id,
@@ -215,6 +675,10 @@ def build_endpoint_plans(
             continue
 
         actions: list[PlanAction] = []
+        planned_retirements: list[
+            RetireRequest
+        ] = []
+
         for identity, desired in sorted(
             desired_by_identity.items()
         ):
@@ -226,23 +690,62 @@ def build_endpoint_plans(
                 (endpoint_id, object_type, logical_key),
                 [],
             )
+            policy = selected_policies.get(object_type)
+
             if len(current_rows) > 1:
-                raise PlanningConflict(
+                selector = (
-                    f"Multiple endpoint objects for {endpoint_id} "
+                    getattr(
-                    f"{object_type} {logical_key!r}"
+                        policy,
+                        "select_current",
+                        None,
+                    )
+                    if policy is not None
+                    else None
                 )
 
-            current = (
+                if selector is None:
-                current_rows[0].object
+                    raise PlanningConflict(
-                if current_rows
+                        f"Multiple endpoint objects for {endpoint_id} "
-                else None
+                        f"{object_type} {logical_key!r}"
-            )
+                    )
+
+                current, retirements = selector(
+                    tuple(current_rows),
+                    desired,
+                )
+
+                planned_retirements.extend(
+                    retirements
+                )
+            else:
+                current = (
+                    current_rows[0].object
+                    if current_rows
+                    else None
+                )
             if current is None:
                 operation = PlanOperation.CREATE
-            elif current.payload == desired.payload:
-                operation = PlanOperation.NOOP
             else:
-                operation = PlanOperation.UPDATE
+                if policy is None:
+                    equivalent = (
+                        current.payload == desired.payload
+                        and current.references == desired.references
+                    )
+                else:
+                    equivalent = policy.equivalent(
+                        current,
+                        desired,
+                    )
+
+                operation = (
+                    PlanOperation.NOOP
+                    if equivalent
+                    else (
+                        policy.mutation_operation
+                        if policy is not None
+                        else PlanOperation.UPDATE
+                    )
+                )
 
             actions.append(
                 PlanAction(
@@ -253,10 +756,101 @@ def build_endpoint_plans(
                 )
             )
 
+        replaced = {
+            (
+                action.desired.object_type,
+                action.desired.logical_key,
+            )
+            for action in actions
+            if (
+                action.operation == PlanOperation.REPLACE
+                and action.desired is not None
+            )
+        }
+
+        retiring_local = {
+            (
+                retirement.object.object_type,
+                retirement.object.local_key,
+            )
+            for retirement in planned_retirements
+        }
+
+        if replaced or retiring_local:
+            promoted: list[PlanAction] = []
+
+            for action in actions:
+                desired = action.desired
+
+                logical_dependency_replaced = (
+                    desired is not None
+                    and any(
+                        (
+                            reference.object_type,
+                            reference.logical_key,
+                        )
+                        in replaced
+                        for reference in desired.references
+                        if reference.logical_key is not None
+                    )
+                )
+
+                raw_rows = (
+                    raw_current_by_identity.get(
+                        (
+                            endpoint_id,
+                            desired.object_type,
+                            desired.logical_key,
+                        ),
+                        [],
+                    )
+                    if desired is not None
+                    else []
+                )
+
+                local_dependency_retiring = any(
+                    (
+                        reference.object_type,
+                        reference.local_key,
+                    )
+                    in retiring_local
+                    for row in raw_rows
+                    for reference in row.object.references
+                    if reference.local_key is not None
+                )
+
+                if (
+                    action.operation == PlanOperation.NOOP
+                    and (
+                        logical_dependency_replaced
+                        or local_dependency_retiring
+                    )
+                ):
+                    reason = (
+                        "referenced object is being replaced"
+                        if logical_dependency_replaced
+                        else (
+                            "referenced local object is being retired"
+                        )
+                    )
+
+                    action = replace(
+                        action,
+                        operation=PlanOperation.UPDATE,
+                        reason=reason,
+                    )
+
+                promoted.append(action)
+
+            actions = promoted
+
         plans.append(
             EndpointPlan(
                 endpoint_id=endpoint_id,
                 actions=order_plan_actions(actions),
+                retirements=tuple(
+                    planned_retirements
+                ),
             )
         )
 

tests/test_cli_reconcile.py

@@ -220,6 +220,7 @@ class ReconcileCliTests(unittest.TestCase):
         execution = SimpleNamespace(
             applied=(),
             skipped=(),
+            retired=(),
         )
 
         with tempfile.TemporaryDirectory() as directory:
@@ -255,11 +256,14 @@ class ReconcileCliTests(unittest.TestCase):
             stdout.getvalue(),
         )
         self.assertIn(
-            "APPLY COMPLETE: 0 action(s) applied and verified",
+            "APPLY COMPLETE: 0 action(s) applied and verified; "
+            "0 retirement(s) completed",
             stdout.getvalue(),
         )
 
-    def test_apply_budget_blocks_executor_before_mutation(self):
+    def test_apply_budget_counts_planned_retirement_before_mutation(
+        self,
+    ):
         config = SimpleNamespace(
             target_ip="192.168.1.101",
             include_disabled=False,
@@ -267,10 +271,18 @@ class ReconcileCliTests(unittest.TestCase):
             domain=None,
         )
         adapter = object()
-        action = SimpleNamespace(
+
-            operation=SimpleNamespace(value="create"),
+        retirement = SimpleNamespace(
+            object=SimpleNamespace(
+                object_type="certificates",
+                logical_key="app.example.com",
+                local_key="63",
+            )
+        )
+        plan = SimpleNamespace(
+            actions=(),
+            retirements=(retirement,),
         )
-        plan = SimpleNamespace(actions=(action,))
         reconciliation = SimpleNamespace(
             observations=(),
             desired=(),
@@ -298,7 +310,64 @@ class ReconcileCliTests(unittest.TestCase):
                 with self.assertRaisesRegex(
                     RuntimeError,
                     (
-                        r"Refusing to apply 1 change\(s\); "
+                        r"Refusing to apply 1 mutation\(s\); "
+                        r"--max-changes allows 0"
+                    ),
+                ):
+                    run_reconcile(
+                        SimpleNamespace(
+                            state_db=str(state_path),
+                            include_disabled=False,
+                            apply=True,
+                            max_changes=0,
+                        )
+                    )
+
+        execute.assert_not_called()
+
+    def test_apply_budget_blocks_executor_before_mutation(self):
+        config = SimpleNamespace(
+            target_ip="192.168.1.101",
+            include_disabled=False,
+            domain_matching=False,
+            domain=None,
+        )
+        adapter = object()
+        action = SimpleNamespace(
+            operation=SimpleNamespace(value="create"),
+        )
+        plan = SimpleNamespace(
+            actions=(action,),
+            retirements=(),
+        )
+        reconciliation = SimpleNamespace(
+            observations=(),
+            desired=(),
+            plans=(plan,),
+            failures=(),
+        )
+
+        with tempfile.TemporaryDirectory() as directory:
+            state_path = Path(directory) / "state.sqlite3"
+
+            with patch(
+                "dns_sync.cli.AppConfig.from_env",
+                return_value=config,
+            ), patch(
+                "dns_sync.cli._configured_adapters",
+                return_value=[adapter],
+            ), patch(
+                "dns_sync.cli.discover_reconcile_plan",
+                return_value=reconciliation,
+            ), patch(
+                "dns_sync.cli.print_endpoint_plans",
+            ), patch(
+                "dns_sync.cli.execute_endpoint_plans",
+            ) as execute, redirect_stdout(io.StringIO()):
+                with self.assertRaisesRegex(
+                    RuntimeError,
+                    (
+                        r"Refusing to apply 1 mutation\(s\); "
                         r"--max-changes allows 0"
                     ),
                 ):

tests/test_executor.py

@@ -16,6 +16,7 @@ from dns_sync.adapters import (
     Observation,
     PlanAction,
     PlanOperation,
+    RetireRequest,
 )
 from dns_sync.executor import execute_endpoint_plans
 from dns_sync.persistence import SQLiteStateStore
@@ -117,6 +118,9 @@ class FakeAdapter:
     events: list[tuple[str, str, str]] = field(
         default_factory=list
     )
+    retired: list[RetireRequest] = field(
+        default_factory=list
+    )
 
     def apply(self, plan_action):
         desired = plan_action.desired
@@ -163,6 +167,16 @@ class FakeAdapter:
             raise self.verify_error
         return self.verify_result
 
+    def retire(self, request):
+        self.events.append(
+            (
+                "retire",
+                request.object.object_type,
+                request.object.logical_key,
+            )
+        )
+        self.retired.append(request)
+
 
 class ExecutionTests(unittest.TestCase):
     def test_successful_actions_are_applied_verified_and_persisted(self):
@@ -216,6 +230,378 @@ class ExecutionTests(unittest.TestCase):
             ("apply", "verify", "apply", "verify"),
         )
 
+    def test_replace_rebinds_dependent_before_retiring_old_object(self):
+        endpoint = Endpoint(
+            "npm",
+            "192.168.1.4:81",
+        )
+
+        class ReplacementAdapter(FakeAdapter):
+            def apply(self, plan_action):
+                result = super().apply(plan_action)
+
+                if (
+                    plan_action.operation
+                    == PlanOperation.REPLACE
+                ):
+                    return ApplyResult(
+                        object=result.object,
+                        detail="replacement",
+                        retirements=(
+                            RetireRequest(
+                                object=plan_action.current,
+                            ),
+                        ),
+                    )
+
+                return result
+
+        adapter = ReplacementAdapter(
+            endpoint,
+            capabilities(
+                CERTIFICATES,
+                PROXY_HOSTS,
+            ),
+            apply_local_keys={
+                (
+                    CERTIFICATES,
+                    "app.example.com",
+                ): "84",
+                (
+                    PROXY_HOSTS,
+                    "app.example.com",
+                ): "12",
+            },
+        )
+
+        old_certificate = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            local_key="67",
+            payload={
+                "domain_names": ["app.example.com"],
+                "valid_from": "2026-08-01T00:00:00+00:00",
+                "expires_on": "2026-11-01T00:00:00+00:00",
+            },
+        )
+
+        desired_certificate = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "valid_from": "2026-09-01T00:00:00+00:00",
+                "expires_on": "2026-12-01T00:00:00+00:00",
+            },
+        )
+
+        old_proxy = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            local_key="12",
+            payload={
+                "domain_names": ["app.example.com"],
+                "forward_scheme": "https",
+                "forward_host": "backend.internal",
+                "forward_port": 443,
+                "enabled": True,
+            },
+            references=(
+                ObjectReference(
+                    relation="certificate",
+                    object_type=CERTIFICATES,
+                    local_key="67",
+                ),
+            ),
+        )
+
+        desired_proxy = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            payload=dict(old_proxy.payload),
+            references=(
+                ObjectReference(
+                    relation="certificate",
+                    object_type=CERTIFICATES,
+                    logical_key="app.example.com",
+                ),
+            ),
+        )
+
+        plan = EndpointPlan(
+            endpoint.identity,
+            (
+                PlanAction(
+                    endpoint_id=endpoint.identity,
+                    operation=PlanOperation.REPLACE,
+                    desired=desired_certificate,
+                    current=old_certificate,
+                ),
+                PlanAction(
+                    endpoint_id=endpoint.identity,
+                    operation=PlanOperation.UPDATE,
+                    desired=desired_proxy,
+                    current=old_proxy,
+                ),
+            ),
+        )
+
+        with SQLiteStateStore(":memory:") as store:
+            store.replace_observations(
+                endpoint.identity,
+                CERTIFICATES,
+                [
+                    Observation(
+                        endpoint.identity,
+                        old_certificate,
+                        NOW,
+                    )
+                ],
+                NOW,
+            )
+
+            store.replace_observations(
+                endpoint.identity,
+                PROXY_HOSTS,
+                [
+                    Observation(
+                        endpoint.identity,
+                        old_proxy,
+                        NOW,
+                    )
+                ],
+                NOW,
+            )
+
+            execute_endpoint_plans(
+                [adapter],
+                [plan],
+                store,
+                executed_at=NOW,
+            )
+
+        self.assertEqual(
+            tuple(event[0] for event in adapter.events),
+            (
+                "apply",
+                "verify",
+                "apply",
+                "verify",
+                "retire",
+            ),
+        )
+
+        proxy_apply = adapter.applied_actions[1]
+
+        self.assertEqual(
+            proxy_apply.desired.references,
+            (
+                ObjectReference(
+                    relation="certificate",
+                    object_type=CERTIFICATES,
+                    local_key="84",
+                ),
+            ),
+        )
+
+        self.assertEqual(
+            adapter.retired,
+            [
+                RetireRequest(
+                    object=old_certificate,
+                )
+            ],
+        )
+
+
+    def test_restart_recovery_rebinds_before_planned_retirement(self):
+        endpoint = Endpoint(
+            "npm",
+            "192.168.1.4:81",
+        )
+
+        adapter = FakeAdapter(
+            endpoint,
+            capabilities(
+                CERTIFICATES,
+                PROXY_HOSTS,
+            ),
+            apply_local_keys={
+                (
+                    PROXY_HOSTS,
+                    "app.example.com",
+                ): "12",
+            },
+        )
+
+        old_certificate = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            local_key="67",
+            payload={
+                "domain_names": ["app.example.com"],
+                "valid_from": "2026-08-01T00:00:00+00:00",
+                "expires_on": "2026-11-01T00:00:00+00:00",
+            },
+        )
+
+        new_certificate = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            local_key="84",
+            payload={
+                "domain_names": ["app.example.com"],
+                "valid_from": "2026-09-01T00:00:00+00:00",
+                "expires_on": "2026-12-01T00:00:00+00:00",
+            },
+        )
+
+        old_proxy = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            local_key="12",
+            payload={
+                "domain_names": ["app.example.com"],
+                "forward_scheme": "https",
+                "forward_host": "backend.internal",
+                "forward_port": 443,
+                "enabled": True,
+            },
+            references=(
+                ObjectReference(
+                    relation="certificate",
+                    object_type=CERTIFICATES,
+                    local_key="67",
+                ),
+            ),
+        )
+
+        desired_certificate = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            payload=dict(new_certificate.payload),
+        )
+
+        desired_proxy = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            payload=dict(old_proxy.payload),
+            references=(
+                ObjectReference(
+                    relation="certificate",
+                    object_type=CERTIFICATES,
+                    logical_key="app.example.com",
+                ),
+            ),
+        )
+
+        plan = EndpointPlan(
+            endpoint.identity,
+            (
+                PlanAction(
+                    endpoint_id=endpoint.identity,
+                    operation=PlanOperation.NOOP,
+                    desired=desired_certificate,
+                    current=new_certificate,
+                ),
+                PlanAction(
+                    endpoint_id=endpoint.identity,
+                    operation=PlanOperation.UPDATE,
+                    desired=desired_proxy,
+                    current=old_proxy,
+                    reason=(
+                        "referenced local object is being retired"
+                    ),
+                ),
+            ),
+            retirements=(
+                RetireRequest(
+                    object=old_certificate,
+                ),
+            ),
+        )
+
+        with SQLiteStateStore(":memory:") as store:
+            store.replace_observations(
+                endpoint.identity,
+                CERTIFICATES,
+                [
+                    Observation(
+                        endpoint.identity,
+                        old_certificate,
+                        NOW,
+                    ),
+                    Observation(
+                        endpoint.identity,
+                        new_certificate,
+                        NOW,
+                    ),
+                ],
+                NOW,
+            )
+
+            store.replace_observations(
+                endpoint.identity,
+                PROXY_HOSTS,
+                [
+                    Observation(
+                        endpoint.identity,
+                        old_proxy,
+                        NOW,
+                    )
+                ],
+                NOW,
+            )
+
+            result = execute_endpoint_plans(
+                [adapter],
+                [plan],
+                store,
+                executed_at=NOW,
+            )
+
+        self.assertEqual(
+            tuple(event[0] for event in adapter.events),
+            (
+                "apply",
+                "verify",
+                "retire",
+            ),
+        )
+
+        proxy_action = adapter.applied_actions[0]
+
+        self.assertEqual(
+            proxy_action.desired.references,
+            (
+                ObjectReference(
+                    relation="certificate",
+                    object_type=CERTIFICATES,
+                    local_key="84",
+                ),
+            ),
+        )
+
+        self.assertEqual(
+            adapter.retired,
+            [
+                RetireRequest(
+                    object=old_certificate,
+                )
+            ],
+        )
+        self.assertEqual(
+            result.retired,
+            (
+                (
+                    endpoint.identity,
+                    RetireRequest(
+                        object=old_certificate,
+                    ),
+                ),
+            ),
+        )
+
     def test_failed_verification_is_persisted_and_raises(self):
         endpoint = Endpoint("verizon", "192.168.1.1:443")
         adapter = FakeAdapter(
@@ -280,6 +666,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

@@ -9,6 +9,8 @@ from dns_sync.adapters import (
     CapabilityOperation,
     NormalizedObject,
     ObjectReference,
+    PlanAction,
+    PlanOperation,
 )
 from dns_sync.certificates.material import CertificateMaterial
 from dns_sync.config import AppConfig
@@ -92,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(
@@ -103,6 +105,10 @@ class NpmAdapterTests(unittest.TestCase):
                         PROXY_HOSTS,
                         CapabilityOperation.DISCOVER,
                     ),
+                    Capability(
+                        PROXY_HOSTS,
+                        CapabilityOperation.APPLY,
+                    ),
                     Capability(
                         PROXY_HOSTS,
                         CapabilityOperation.VERIFY,
@@ -111,6 +117,10 @@ class NpmAdapterTests(unittest.TestCase):
                         CERTIFICATES,
                         CapabilityOperation.DISCOVER,
                     ),
+                    Capability(
+                        CERTIFICATES,
+                        CapabilityOperation.APPLY,
+                    ),
                     Capability(
                         CERTIFICATES,
                         CapabilityOperation.VERIFY,
@@ -119,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 = [
@@ -762,14 +757,862 @@ class NpmAdapterTests(unittest.TestCase):
 
         self.assertEqual(resolved, material)
 
+    def test_certificate_create_uploads_verifies_and_returns_target_id(self):
+        target_client = Mock()
+        source_client = Mock()
+
+        source_client.certificates.return_value = [
+            {
+                "id": 10,
+                "domain_names": ["app.example.com"],
+                "provider": "letsencrypt",
+                "nice_name": "source",
+                "valid_from": "2026-09-01T00:00:00Z",
+                "expires_on": "2026-12-01T00:00:00Z",
+                "meta": {},
+            }
+        ]
+
+        source_client.certificate_material.return_value = (
+            CertificateMaterial(
+                certificate=b"CERT",
+                certificate_key=b"KEY",
+                intermediate_certificate=b"CHAIN",
+            )
+        )
+
+        target_client.create_certificate.return_value = {
+            "id": 84,
+        }
+
+        target_client.certificate.return_value = {
+            "id": 84,
+            "domain_names": ["app.example.com"],
+            "provider": "other",
+            "nice_name": "source",
+            "valid_from": "2026-09-01T00:00:00Z",
+            "expires_on": "2026-12-01T00:00:00Z",
+            "meta": {
+                "cluster_managed": True,
+                "acme_renew": False,
+                "dns_challenge": False,
+            },
+        }
+
+        adapter, client_factory = (
+            self.target_adapter_with_sources(
+                target_client,
+                {
+                    "npm://192.168.1.3:81": source_client,
+                },
+            )
+        )
+
+        desired = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "provider": "letsencrypt",
+                "nice_name": "source",
+                "valid_from": "2026-09-01T00:00:00+00:00",
+                "expires_on": "2026-12-01T00:00:00+00:00",
+                "cluster_managed": True,
+                "acme_renew": True,
+                "dns_challenge": False,
+            },
+        )
+
+        action = PlanAction(
+            endpoint_id=adapter.endpoint.identity,
+            operation=PlanOperation.CREATE,
+            desired=desired,
+        )
+
+        with patch(
+            "dns_sync.plugins.adapters.npm.NpmClient",
+            side_effect=client_factory,
+        ):
+            result = adapter.apply(action)
+
+        self.assertEqual(
+            result.object.local_key,
+            "84",
+        )
+        self.assertEqual(
+            result.object.logical_key,
+            "app.example.com",
+        )
+
+        target_client.create_certificate.assert_called_once_with(
+            {
+                "provider": "other",
+                "domain_names": ["app.example.com"],
+                "nice_name": "source",
+                "meta": {
+                    "cluster_managed": True,
+                    "acme_renew": False,
+                    "dns_challenge": False,
+                },
+            }
+        )
+
+        target_client.upload_certificate.assert_called_once_with(
+            "84",
+            certificate=b"CERT",
+            certificate_key=b"KEY",
+            intermediate_certificate=b"CHAIN",
+        )
+
+        target_client.certificate.assert_called_once_with("84")
+        target_client.delete_certificate.assert_not_called()
+
+    def test_certificate_create_rolls_back_when_upload_fails(self):
+        target_client = Mock()
+        source_client = Mock()
+
+        source_client.certificates.return_value = [
+            {
+                "id": 10,
+                "domain_names": ["app.example.com"],
+                "provider": "other",
+                "valid_from": "2026-09-01T00:00:00Z",
+                "expires_on": "2026-12-01T00:00:00Z",
+                "meta": {},
+            }
+        ]
+
+        source_client.certificate_material.return_value = (
+            CertificateMaterial(
+                certificate=b"CERT",
+                certificate_key=b"KEY",
+            )
+        )
+
+        target_client.create_certificate.return_value = {
+            "id": 84,
+        }
+        target_client.upload_certificate.side_effect = RuntimeError(
+            "upload failed"
+        )
+
+        adapter, client_factory = (
+            self.target_adapter_with_sources(
+                target_client,
+                {
+                    "npm://192.168.1.3:81": source_client,
+                },
+            )
+        )
+
+        desired = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "valid_from": "2026-09-01T00:00:00+00:00",
+                "expires_on": "2026-12-01T00:00:00+00:00",
+            },
+        )
+
+        action = PlanAction(
+            endpoint_id=adapter.endpoint.identity,
+            operation=PlanOperation.CREATE,
+            desired=desired,
+        )
+
+        with patch(
+            "dns_sync.plugins.adapters.npm.NpmClient",
+            side_effect=client_factory,
+        ):
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "upload failed",
+            ):
+                adapter.apply(action)
+
+        target_client.delete_certificate.assert_called_once_with(
+            "84"
+        )
+        target_client.certificate.assert_not_called()
+
+    def test_certificate_create_rolls_back_when_created_state_is_wrong(self):
+        target_client = Mock()
+        source_client = Mock()
+
+        source_client.certificates.return_value = [
+            {
+                "id": 10,
+                "domain_names": ["app.example.com"],
+                "provider": "other",
+                "valid_from": "2026-09-01T00:00:00Z",
+                "expires_on": "2026-12-01T00:00:00Z",
+                "meta": {},
+            }
+        ]
+
+        source_client.certificate_material.return_value = (
+            CertificateMaterial(
+                certificate=b"CERT",
+                certificate_key=b"KEY",
+            )
+        )
+
+        target_client.create_certificate.return_value = {
+            "id": 84,
+        }
+
+        target_client.certificate.return_value = {
+            "id": 84,
+            "domain_names": ["app.example.com"],
+            "provider": "other",
+            "valid_from": "2026-09-01T00:00:00Z",
+            # Wrong target certificate state.
+            "expires_on": "2027-01-01T00:00:00Z",
+            "meta": {},
+        }
+
+        adapter, client_factory = (
+            self.target_adapter_with_sources(
+                target_client,
+                {
+                    "npm://192.168.1.3:81": source_client,
+                },
+            )
+        )
+
+        desired = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "valid_from": "2026-09-01T00:00:00+00:00",
+                "expires_on": "2026-12-01T00:00:00+00:00",
+            },
+        )
+
+        action = PlanAction(
+            endpoint_id=adapter.endpoint.identity,
+            operation=PlanOperation.CREATE,
+            desired=desired,
+        )
+
+        with patch(
+            "dns_sync.plugins.adapters.npm.NpmClient",
+            side_effect=client_factory,
+        ):
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "does not match desired",
+            ):
+                adapter.apply(action)
+
+        target_client.delete_certificate.assert_called_once_with(
+            "84"
+        )
+
+    def test_certificate_create_reports_rollback_failure_without_secrets(self):
+        target_client = Mock()
+        source_client = Mock()
+
+        source_client.certificates.return_value = [
+            {
+                "id": 10,
+                "domain_names": ["app.example.com"],
+                "provider": "other",
+                "valid_from": "2026-09-01T00:00:00Z",
+                "expires_on": "2026-12-01T00:00:00Z",
+                "meta": {},
+            }
+        ]
+
+        source_client.certificate_material.return_value = (
+            CertificateMaterial(
+                certificate=b"DO-NOT-PRINT-CERT",
+                certificate_key=b"DO-NOT-PRINT-KEY",
+            )
+        )
+
+        target_client.create_certificate.return_value = {
+            "id": 84,
+        }
+        target_client.upload_certificate.side_effect = RuntimeError(
+            "upload failed"
+        )
+        target_client.delete_certificate.side_effect = RuntimeError(
+            "delete failed"
+        )
+
+        adapter, client_factory = (
+            self.target_adapter_with_sources(
+                target_client,
+                {
+                    "npm://192.168.1.3:81": source_client,
+                },
+            )
+        )
+
+        desired = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "valid_from": "2026-09-01T00:00:00+00:00",
+                "expires_on": "2026-12-01T00:00:00+00:00",
+            },
+        )
+
+        action = PlanAction(
+            endpoint_id=adapter.endpoint.identity,
+            operation=PlanOperation.CREATE,
+            desired=desired,
+        )
+
+        with patch(
+            "dns_sync.plugins.adapters.npm.NpmClient",
+            side_effect=client_factory,
+        ):
+            with self.assertRaises(RuntimeError) as caught:
+                adapter.apply(action)
+
+        rendered = str(caught.exception)
+
+        self.assertIn(
+            "rollback also failed",
+            rendered,
+        )
+        self.assertNotIn(
+            "DO-NOT-PRINT-CERT",
+            rendered,
+        )
+        self.assertNotIn(
+            "DO-NOT-PRINT-KEY",
+            rendered,
+        )
+
+    def test_certificate_create_rejects_invalid_created_id(self):
+        target_client = Mock()
+        source_client = Mock()
+
+        source_client.certificates.return_value = [
+            {
+                "id": 10,
+                "domain_names": ["app.example.com"],
+                "provider": "other",
+                "valid_from": "2026-09-01T00:00:00Z",
+                "expires_on": "2026-12-01T00:00:00Z",
+                "meta": {},
+            }
+        ]
+
+        source_client.certificate_material.return_value = (
+            CertificateMaterial(
+                certificate=b"CERT",
+                certificate_key=b"KEY",
+            )
+        )
+
+        target_client.create_certificate.return_value = {
+            "id": None,
+        }
+
+        adapter, client_factory = (
+            self.target_adapter_with_sources(
+                target_client,
+                {
+                    "npm://192.168.1.3:81": source_client,
+                },
+            )
+        )
+
+        desired = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "valid_from": "2026-09-01T00:00:00+00:00",
+                "expires_on": "2026-12-01T00:00:00+00:00",
+            },
+        )
+
+        action = PlanAction(
+            endpoint_id=adapter.endpoint.identity,
+            operation=PlanOperation.CREATE,
+            desired=desired,
+        )
+
+        with patch(
+            "dns_sync.plugins.adapters.npm.NpmClient",
+            side_effect=client_factory,
+        ):
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "invalid id",
+            ):
+                adapter.apply(action)
+
+        target_client.upload_certificate.assert_not_called()
+        target_client.delete_certificate.assert_not_called()
+
+    def test_proxy_host_create_uses_resolved_target_certificate_id(self):
+        client = Mock()
+
+        client.create_proxy_host.return_value = {
+            "id": 120,
+        }
+
+        client.proxy_host.return_value = {
+            "id": 120,
+            "domain_names": ["app.example.com"],
+            "forward_scheme": "http",
+            "forward_host": "backend.internal",
+            "forward_port": 8080,
+            "certificate_id": 84,
+            "ssl_forced": True,
+            "caching_enabled": False,
+            "block_exploits": True,
+            "allow_websocket_upgrade": True,
+            "http2_support": True,
+            "hsts_enabled": True,
+            "hsts_subdomains": False,
+            "enabled": True,
+        }
+
+        adapter = self.adapter(client)
+
+        desired = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "forward_scheme": "http",
+                "forward_host": "backend.internal",
+                "forward_port": 8080,
+                "ssl_forced": True,
+                "caching_enabled": False,
+                "block_exploits": True,
+                "allow_websocket_upgrade": True,
+                "http2_support": True,
+                "hsts_enabled": True,
+                "hsts_subdomains": False,
+                "enabled": True,
+            },
+            references=(
+                ObjectReference(
+                    relation="certificate",
+                    object_type=CERTIFICATES,
+                    local_key="84",
+                ),
+            ),
+        )
+
+        result = adapter.apply(
+            PlanAction(
+                endpoint_id=adapter.endpoint.identity,
+                operation=PlanOperation.CREATE,
+                desired=desired,
+            )
+        )
+
+        client.create_proxy_host.assert_called_once_with(
+            {
+                "domain_names": ["app.example.com"],
+                "forward_scheme": "http",
+                "forward_host": "backend.internal",
+                "forward_port": 8080,
+                "certificate_id": 84,
+                "ssl_forced": True,
+                "caching_enabled": False,
+                "block_exploits": True,
+                "allow_websocket_upgrade": True,
+                "http2_support": True,
+                "hsts_enabled": True,
+                "hsts_subdomains": False,
+                "enabled": True,
+            }
+        )
+
+        self.assertEqual(
+            result.object.local_key,
+            "120",
+        )
+        self.assertEqual(
+            result.object.references,
+            desired.references,
+        )
+        client.delete_proxy_host.assert_not_called()
+
+    def test_proxy_host_create_without_certificate_uses_zero(self):
+        client = Mock()
+
+        client.create_proxy_host.return_value = {
+            "id": 120,
+        }
+
+        client.proxy_host.return_value = {
+            "id": 120,
+            "domain_names": ["plain.example.com"],
+            "forward_scheme": "http",
+            "forward_host": "backend.internal",
+            "forward_port": 8080,
+            "certificate_id": 0,
+            "enabled": True,
+        }
+
+        adapter = self.adapter(client)
+
+        desired = normalize_discovered_object(
+            adapter._proxy_host_from_row(
+                client.proxy_host.return_value
+            )
+        )
+
+        desired = NormalizedObject(
+            object_type=desired.object_type,
+            logical_key=desired.logical_key,
+            payload=desired.payload,
+            references=desired.references,
+        )
+
+        result = adapter.apply(
+            PlanAction(
+                endpoint_id=adapter.endpoint.identity,
+                operation=PlanOperation.CREATE,
+                desired=desired,
+            )
+        )
+
+        payload = client.create_proxy_host.call_args.kwargs
+        if not payload:
+            payload = client.create_proxy_host.call_args.args[0]
+
+        self.assertEqual(
+            payload["certificate_id"],
+            0,
+        )
+        self.assertEqual(
+            result.object.references,
+            (),
+        )
+
+    def test_proxy_host_create_rolls_back_when_refetch_mismatches(self):
+        client = Mock()
+
+        client.create_proxy_host.return_value = {
+            "id": 120,
+        }
+
+        client.proxy_host.return_value = {
+            "id": 120,
+            "domain_names": ["wrong.example.com"],
+            "forward_scheme": "http",
+            "forward_host": "backend.internal",
+            "forward_port": 8080,
+            "certificate_id": 0,
+            "enabled": True,
+        }
+
+        adapter = self.adapter(client)
+
+        desired = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "forward_scheme": "http",
+                "forward_host": "backend.internal",
+                "forward_port": 8080,
+                "ssl_forced": False,
+                "caching_enabled": False,
+                "block_exploits": False,
+                "allow_websocket_upgrade": False,
+                "http2_support": False,
+                "hsts_enabled": False,
+                "hsts_subdomains": False,
+                "enabled": True,
+            },
+        )
+
+        with self.assertRaisesRegex(
+            RuntimeError,
+            "does not match desired",
+        ):
+            adapter.apply(
+                PlanAction(
+                    endpoint_id=adapter.endpoint.identity,
+                    operation=PlanOperation.CREATE,
+                    desired=desired,
+                )
+            )
+
+        client.delete_proxy_host.assert_called_once_with(
+            "120"
+        )
+
+    def test_proxy_host_create_reports_rollback_failure(self):
+        client = Mock()
+
+        client.create_proxy_host.return_value = {
+            "id": 120,
+        }
+        client.proxy_host.side_effect = RuntimeError(
+            "refetch failed"
+        )
+        client.delete_proxy_host.side_effect = RuntimeError(
+            "delete failed"
+        )
+
+        adapter = self.adapter(client)
+
+        desired = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "forward_scheme": "http",
+                "forward_host": "backend.internal",
+                "forward_port": 8080,
+            },
+        )
+
+        with self.assertRaisesRegex(
+            RuntimeError,
+            "rollback also failed",
+        ):
+            adapter.apply(
+                PlanAction(
+                    endpoint_id=adapter.endpoint.identity,
+                    operation=PlanOperation.CREATE,
+                    desired=desired,
+                )
+            )
+
+    def test_proxy_host_update_uses_current_local_id_and_resolved_reference(self):
+        client = Mock()
+
+        client.proxy_host.return_value = {
+            "id": 12,
+            "domain_names": ["app.example.com"],
+            "forward_scheme": "https",
+            "forward_host": "new-backend.internal",
+            "forward_port": 8443,
+            "certificate_id": 84,
+            "ssl_forced": True,
+            "enabled": True,
+        }
+
+        adapter = self.adapter(client)
+
+        current = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            local_key="12",
+            payload={
+                "domain_names": ["app.example.com"],
+                "forward_scheme": "http",
+                "forward_host": "old-backend.internal",
+                "forward_port": 8080,
+                "ssl_forced": False,
+                "caching_enabled": False,
+                "block_exploits": False,
+                "allow_websocket_upgrade": False,
+                "http2_support": False,
+                "hsts_enabled": False,
+                "hsts_subdomains": False,
+                "enabled": True,
+            },
+        )
+
+        actual = normalize_discovered_object(
+            adapter._proxy_host_from_row(
+                client.proxy_host.return_value
+            )
+        )
+
+        desired = NormalizedObject(
+            object_type=actual.object_type,
+            logical_key=actual.logical_key,
+            payload=actual.payload,
+            references=actual.references,
+        )
+
+        result = adapter.apply(
+            PlanAction(
+                endpoint_id=adapter.endpoint.identity,
+                operation=PlanOperation.UPDATE,
+                desired=desired,
+                current=current,
+            )
+        )
+
+        client.update_proxy_host.assert_called_once()
+
+        args = client.update_proxy_host.call_args.args
+        self.assertEqual(args[0], "12")
+        self.assertEqual(
+            args[1]["certificate_id"],
+            84,
+        )
+
+        self.assertEqual(
+            result.object.local_key,
+            "12",
+        )
+        self.assertEqual(
+            result.object.references,
+            desired.references,
+        )
+        client.delete_proxy_host.assert_not_called()
+
+    def test_proxy_host_update_fails_when_target_does_not_converge(self):
+        client = Mock()
+
+        client.proxy_host.return_value = {
+            "id": 12,
+            "domain_names": ["app.example.com"],
+            "forward_scheme": "http",
+            "forward_host": "still-old.internal",
+            "forward_port": 8080,
+            "certificate_id": 0,
+            "enabled": True,
+        }
+
+        adapter = self.adapter(client)
+
+        current = normalize_discovered_object(
+            adapter._proxy_host_from_row(
+                client.proxy_host.return_value
+            )
+        )
+
+        desired = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "forward_scheme": "http",
+                "forward_host": "new.internal",
+                "forward_port": 8080,
+                "ssl_forced": False,
+                "caching_enabled": False,
+                "block_exploits": False,
+                "allow_websocket_upgrade": False,
+                "http2_support": False,
+                "hsts_enabled": False,
+                "hsts_subdomains": False,
+                "enabled": True,
+            },
+        )
+
+        with self.assertRaisesRegex(
+            RuntimeError,
+            "does not match desired",
+        ):
+            adapter.apply(
+                PlanAction(
+                    endpoint_id=adapter.endpoint.identity,
+                    operation=PlanOperation.UPDATE,
+                    desired=desired,
+                    current=current,
+                )
+            )
+
+        client.delete_proxy_host.assert_not_called()
+
+    def test_proxy_host_apply_rejects_unresolved_certificate_reference(self):
+        client = Mock()
+        adapter = self.adapter(client)
+
+        desired = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "forward_scheme": "http",
+                "forward_host": "backend.internal",
+                "forward_port": 8080,
+            },
+            references=(
+                ObjectReference(
+                    relation="certificate",
+                    object_type=CERTIFICATES,
+                    logical_key="app.example.com",
+                ),
+            ),
+        )
+
+        with self.assertRaisesRegex(
+            RuntimeError,
+            "not resolved",
+        ):
+            adapter.apply(
+                PlanAction(
+                    endpoint_id=adapter.endpoint.identity,
+                    operation=PlanOperation.CREATE,
+                    desired=desired,
+                )
+            )
+
+        client.create_proxy_host.assert_not_called()
+        client.update_proxy_host.assert_not_called()
+
+    def test_proxy_host_apply_rejects_unsupported_reference(self):
+        client = Mock()
+        adapter = self.adapter(client)
+
+        desired = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "forward_scheme": "http",
+                "forward_host": "backend.internal",
+                "forward_port": 8080,
+            },
+            references=(
+                ObjectReference(
+                    relation="other",
+                    object_type=CERTIFICATES,
+                    local_key="84",
+                ),
+            ),
+        )
+
+        with self.assertRaisesRegex(
+            RuntimeError,
+            "unsupported object reference",
+        ):
+            adapter.apply(
+                PlanAction(
+                    endpoint_id=adapter.endpoint.identity,
+                    operation=PlanOperation.CREATE,
+                    desired=desired,
+                )
+            )
+
+        client.create_proxy_host.assert_not_called()
+
     def test_unsupported_operations_fail_explicitly(self):
         adapter = self.adapter(Mock())
 
         with self.assertRaisesRegex(RuntimeError, "cannot discover"):
             adapter.discover("unsupported_objects")
 
-        with self.assertRaisesRegex(RuntimeError, "does not support apply"):
+        unsupported_action = PlanAction(
-            adapter.apply(Mock())
+            endpoint_id=adapter.endpoint.identity,
+            operation=PlanOperation.CREATE,
+            desired=NormalizedObject(
+                object_type="unsupported_objects",
+                logical_key="unsupported",
+                payload={},
+            ),
+        )
+
+        with self.assertRaisesRegex(
+            RuntimeError,
+            "does not support",
+        ):
+            adapter.apply(unsupported_action)
 
         unsupported = NormalizedObject(
             object_type="unsupported_objects",

tests/test_npm_client.py

@@ -186,6 +186,19 @@ class NpmClientMutationTests(unittest.TestCase):
         )
         self.assertEqual(result["id"], 12)
 
+    def test_deletes_proxy_host_by_endpoint_local_id(self):
+        client = self.client()
+        client.request = Mock()
+
+        result = client.delete_proxy_host("12")
+
+        self.assertIsNone(result)
+        client.request.assert_called_once_with(
+            "DELETE",
+            "/nginx/proxy-hosts/12",
+            expected_status=200,
+        )
+
     def test_gets_one_certificate_by_endpoint_local_id(self):
         client = self.client()
         client.get_json = Mock(
@@ -321,6 +334,19 @@ class NpmClientMutationTests(unittest.TestCase):
         ):
             client.certificate_material("99")
 
+    def test_deletes_certificate_by_endpoint_local_id(self):
+        client = self.client()
+        client.request = Mock()
+
+        result = client.delete_certificate("84")
+
+        self.assertIsNone(result)
+        client.request.assert_called_once_with(
+            "DELETE",
+            "/nginx/certificates/84",
+            expected_status=200,
+        )
+
     def test_creates_certificate_metadata_with_json_payload(self):
         client = self.client()
         payload = {

tests/test_orchestrator.py

@@ -147,9 +147,19 @@ class ReadOnlyOrchestrationTests(unittest.TestCase):
                 DNS_RECORDS
             )
 
-        self.assertEqual(len(result.desired), 1)
+        self.assertEqual(len(
+                [
+                    item
+                    for item in result.desired
+                    if item.object_type == DNS_RECORDS
+                ]
+            ), 1)
         self.assertEqual(
-            result.desired[0].payload["ip"],
+            next(
+                item
+                for item in result.desired
+                if item.object_type == DNS_RECORDS
+            ).payload["ip"],
             "192.168.1.101",
         )
         self.assertEqual(len(result.plans), 1)
@@ -220,12 +230,20 @@ class ReadOnlyOrchestrationTests(unittest.TestCase):
             )
 
         self.assertEqual(
-            tuple(item.logical_key for item in result.desired),
+            tuple(
-            ("cached.example.com",),
+                item.logical_key
+                for item in result.desired
+                if item.object_type == DNS_RECORDS
+            ),
+            (),
         )
         self.assertEqual(
-            result.plans[0].actions[0].operation,
+            len(result.plans),
-            PlanOperation.CREATE,
+            1,
+        )
+        self.assertEqual(
+            result.plans[0].actions,
+            (),
         )
         self.assertFalse(status.is_live)
         self.assertEqual(status.last_success_at, T1)
@@ -304,7 +322,11 @@ class ReadOnlyOrchestrationTests(unittest.TestCase):
             )
 
         self.assertEqual(
-            tuple(item.logical_key for item in result.desired),
+            tuple(
+                item.logical_key
+                for item in result.desired
+                if item.object_type == DNS_RECORDS
+            ),
             ("app.example.com",),
         )
 
@@ -344,7 +366,11 @@ class ReadOnlyOrchestrationTests(unittest.TestCase):
             )
 
         self.assertEqual(
-            tuple(item.logical_key for item in result.desired),
+            tuple(
+                item.logical_key
+                for item in result.desired
+                if item.object_type == DNS_RECORDS
+            ),
             ("active.example.com",),
         )
 

tests/test_orchestrator_authority.py

@@ -0,0 +1,581 @@
+import unittest
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+
+from dns_sync.adapters import (
+    CERTIFICATES,
+    DNS_RECORDS,
+    PROXY_HOSTS,
+    Capability,
+    CapabilityOperation,
+    DiscoveredObject,
+    Endpoint,
+    NormalizedObject,
+    ObjectReference,
+    Observation,
+)
+from dns_sync.orchestrator import discover_reconcile_plan
+from dns_sync.reconciliation import ReconciliationConflict
+from dns_sync.persistence import SQLiteStateStore
+
+
+T1 = datetime(
+    2026, 9, 20, 8, 0, tzinfo=timezone.utc
+)
+T2 = datetime(
+    2026, 9, 20, 9, 0, tzinfo=timezone.utc
+)
+
+
+@dataclass
+class FakeAdapter:
+    endpoint: Endpoint
+    capabilities: frozenset[Capability]
+    discovered: dict[str, list[DiscoveredObject]] = field(
+        default_factory=dict
+    )
+    failures: dict[str, str] = field(
+        default_factory=dict
+    )
+
+    def connect(self) -> None:
+        return None
+
+    def discover(
+        self,
+        object_type: str,
+    ) -> list[DiscoveredObject]:
+        if object_type in self.failures:
+            raise RuntimeError(
+                self.failures[object_type]
+            )
+        return list(
+            self.discovered.get(
+                object_type,
+                [],
+            )
+        )
+
+
+def capability_set(
+    *object_types: str,
+    apply: bool = False,
+) -> frozenset[Capability]:
+    rows = {
+        Capability(
+            object_type,
+            CapabilityOperation.DISCOVER,
+        )
+        for object_type in object_types
+    }
+
+    if apply:
+        rows.update(
+            Capability(
+                object_type,
+                CapabilityOperation.APPLY,
+            )
+            for object_type in object_types
+        )
+
+    return frozenset(rows)
+
+
+def normalized_certificate(
+    local_key: str,
+    *,
+    valid_from: str,
+    expires_on: str,
+) -> NormalizedObject:
+    return NormalizedObject(
+        object_type=CERTIFICATES,
+        logical_key="app.example.com",
+        local_key=local_key,
+        payload={
+            "domain_names": ["app.example.com"],
+            "provider": "other",
+            "valid_from": valid_from,
+            "expires_on": expires_on,
+        },
+    )
+
+
+def discovered_certificate(
+    local_key: str,
+    *,
+    valid_from: str,
+    expires_on: str,
+) -> DiscoveredObject:
+    return DiscoveredObject(
+        object_type=CERTIFICATES,
+        local_key=local_key,
+        payload={
+            "domain_names": ["app.example.com"],
+            "provider": "other",
+            "valid_from": valid_from,
+            "expires_on": expires_on,
+        },
+    )
+
+
+def normalized_proxy(
+    local_key: str,
+) -> NormalizedObject:
+    return NormalizedObject(
+        object_type=PROXY_HOSTS,
+        logical_key="app.example.com",
+        local_key=local_key,
+        payload={
+            "domain_names": ["app.example.com"],
+            "forward_scheme": "http",
+            "forward_host": "backend.internal",
+            "forward_port": 8080,
+            "enabled": True,
+        },
+    )
+
+
+class ObservationAuthorityTests(unittest.TestCase):
+    def test_failed_newer_certificate_cache_cannot_override_live_certificate(self):
+        live_endpoint = Endpoint(
+            "npm",
+            "192.168.1.3:81",
+        )
+        failed_endpoint = Endpoint(
+            "npm",
+            "192.168.1.4:81",
+        )
+
+        live = FakeAdapter(
+            endpoint=live_endpoint,
+            capabilities=capability_set(
+                CERTIFICATES,
+            ),
+            discovered={
+                CERTIFICATES: [
+                    discovered_certificate(
+                        "20",
+                        valid_from=(
+                            "2026-09-15T00:00:00+00:00"
+                        ),
+                        expires_on=(
+                            "2026-12-15T00:00:00+00:00"
+                        ),
+                    )
+                ]
+            },
+        )
+
+        failed = FakeAdapter(
+            endpoint=failed_endpoint,
+            capabilities=capability_set(
+                CERTIFICATES,
+                apply=True,
+            ),
+            failures={
+                CERTIFICATES: "offline"
+            },
+        )
+
+        cached_newer = normalized_certificate(
+            "99",
+            valid_from=(
+                "2026-09-19T00:00:00+00:00"
+            ),
+            expires_on=(
+                "2026-12-19T00:00:00+00:00"
+            ),
+        )
+
+        with SQLiteStateStore(":memory:") as store:
+            store.replace_observations(
+                failed_endpoint.identity,
+                CERTIFICATES,
+                [
+                    Observation(
+                        failed_endpoint.identity,
+                        cached_newer,
+                        T1,
+                    )
+                ],
+                T1,
+            )
+
+            result = discover_reconcile_plan(
+                [
+                    live,
+                    failed,
+                ],
+                store,
+                target_ip="192.168.1.101",
+                run_at=T2,
+            )
+
+        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-15T00:00:00+00:00",
+        )
+
+        # The failed target advertised APPLY, but it must not
+        # receive a certificate mutation plan this run.
+        self.assertEqual(
+            result.plans,
+            (),
+        )
+
+        self.assertTrue(
+            any(
+                observation.endpoint_id
+                == failed_endpoint.identity
+                and observation.object.local_key == "99"
+                for observation in result.observations
+            )
+        )
+
+    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_expired_certificate_on_disabled_proxy_is_removed_from_desired(
+        self,
+    ):
+        endpoint = Endpoint(
+            "npm",
+            "192.168.1.3:81",
+        )
+
+        adapter = FakeAdapter(
+            endpoint=endpoint,
+            capabilities=capability_set(
+                CERTIFICATES,
+                PROXY_HOSTS,
+            ),
+            discovered={
+                CERTIFICATES: [
+                    DiscoveredObject(
+                        object_type=CERTIFICATES,
+                        local_key="56",
+                        payload={
+                            "domain_names": [
+                                "admin.trunavigate.org"
+                            ],
+                            "provider": "letsencrypt",
+                            "valid_from": None,
+                            "expires_on": (
+                                "2026-06-22T03:40:57.000Z"
+                            ),
+                        },
+                    )
+                ],
+                PROXY_HOSTS: [
+                    DiscoveredObject(
+                        object_type=PROXY_HOSTS,
+                        local_key="29",
+                        payload={
+                            "domain_names": [
+                                "admin.trunavigate.org"
+                            ],
+                            "forward_scheme": "http",
+                            "forward_host": "192.168.1.4",
+                            "forward_port": 3283,
+                            "enabled": False,
+                        },
+                        references=(
+                            ObjectReference(
+                                relation="certificate",
+                                object_type=CERTIFICATES,
+                                local_key="56",
+                            ),
+                        ),
+                    )
+                ],
+            },
+        )
+
+        with SQLiteStateStore(":memory:") as store:
+            result = discover_reconcile_plan(
+                [adapter],
+                store,
+                target_ip="192.168.1.101",
+                run_at=T2,
+            )
+
+        certificates = [
+            object
+            for object in result.desired
+            if object.object_type == CERTIFICATES
+        ]
+        proxies = [
+            object
+            for object in result.desired
+            if object.object_type == PROXY_HOSTS
+        ]
+
+        self.assertEqual(certificates, [])
+        self.assertEqual(len(proxies), 1)
+        self.assertEqual(
+            proxies[0].logical_key,
+            "admin.trunavigate.org",
+        )
+        self.assertEqual(
+            proxies[0].references,
+            (),
+        )
+
+    def test_expired_certificate_on_enabled_proxy_fails_closed(
+        self,
+    ):
+        endpoint = Endpoint(
+            "npm",
+            "192.168.1.3:81",
+        )
+
+        adapter = FakeAdapter(
+            endpoint=endpoint,
+            capabilities=capability_set(
+                CERTIFICATES,
+                PROXY_HOSTS,
+            ),
+            discovered={
+                CERTIFICATES: [
+                    DiscoveredObject(
+                        object_type=CERTIFICATES,
+                        local_key="56",
+                        payload={
+                            "domain_names": [
+                                "active.example.com"
+                            ],
+                            "provider": "letsencrypt",
+                            "valid_from": None,
+                            "expires_on": (
+                                "2026-06-22T03:40:57.000Z"
+                            ),
+                        },
+                    )
+                ],
+                PROXY_HOSTS: [
+                    DiscoveredObject(
+                        object_type=PROXY_HOSTS,
+                        local_key="29",
+                        payload={
+                            "domain_names": [
+                                "active.example.com"
+                            ],
+                            "forward_scheme": "http",
+                            "forward_host": "192.168.1.4",
+                            "forward_port": 3283,
+                            "enabled": True,
+                        },
+                        references=(
+                            ObjectReference(
+                                relation="certificate",
+                                object_type=CERTIFICATES,
+                                local_key="56",
+                            ),
+                        ),
+                    )
+                ],
+            },
+        )
+
+        with SQLiteStateStore(":memory:") as store:
+            with self.assertRaisesRegex(
+                ReconciliationConflict,
+                "Enabled proxy host references an expired certificate",
+            ):
+                discover_reconcile_plan(
+                    [adapter],
+                    store,
+                    target_ip="192.168.1.101",
+                    run_at=T2,
+                )
+
+    def test_all_failed_sources_preserve_persisted_desired_without_planning(self):
+        npm_endpoint = Endpoint(
+            "npm",
+            "192.168.1.3:81",
+        )
+
+        npm = FakeAdapter(
+            endpoint=npm_endpoint,
+            capabilities=capability_set(
+                PROXY_HOSTS,
+                apply=True,
+            ),
+            failures={
+                PROXY_HOSTS: "offline"
+            },
+        )
+
+        cached_proxy = normalized_proxy("12")
+
+        previous_dns = NormalizedObject(
+            object_type=DNS_RECORDS,
+            logical_key="app.example.com",
+            payload={
+                "hostname": "app.example.com",
+                "ip_address": "192.168.1.101",
+            },
+        )
+
+        with SQLiteStateStore(":memory:") as store:
+            store.replace_observations(
+                npm_endpoint.identity,
+                PROXY_HOSTS,
+                [
+                    Observation(
+                        npm_endpoint.identity,
+                        cached_proxy,
+                        T1,
+                    )
+                ],
+                T1,
+            )
+
+            store.replace_desired_state(
+                DNS_RECORDS,
+                [previous_dns],
+                T1,
+            )
+
+            result = discover_reconcile_plan(
+                [npm],
+                store,
+                target_ip="192.168.1.101",
+                run_at=T2,
+            )
+
+            persisted_dns = store.load_desired_state(
+                DNS_RECORDS
+            )
+
+        self.assertEqual(
+            result.desired,
+            (),
+        )
+        self.assertEqual(
+            result.plans,
+            (),
+        )
+
+        self.assertEqual(
+            len(persisted_dns),
+            1,
+        )
+        self.assertEqual(
+            persisted_dns[0].object,
+            previous_dns,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_orchestrator_npm_desired.py

@@ -0,0 +1,318 @@
+import unittest
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from dns_sync.adapters import (
+    CERTIFICATES,
+    DNS_RECORDS,
+    PROXY_HOSTS,
+    Capability,
+    CapabilityOperation,
+    DiscoveredObject,
+    Endpoint,
+    ObjectReference,
+)
+from dns_sync.orchestrator import discover_reconcile_plan
+from dns_sync.persistence import SQLiteStateStore
+
+
+NOW = datetime(
+    2026,
+    9,
+    20,
+    9,
+    30,
+    tzinfo=timezone.utc,
+)
+
+
+@dataclass
+class FakeNpmAdapter:
+    endpoint: Endpoint
+
+    @property
+    def capabilities(self):
+        return frozenset(
+            {
+                Capability(
+                    CERTIFICATES,
+                    CapabilityOperation.DISCOVER,
+                ),
+                Capability(
+                    CERTIFICATES,
+                    CapabilityOperation.VERIFY,
+                ),
+                Capability(
+                    PROXY_HOSTS,
+                    CapabilityOperation.DISCOVER,
+                ),
+                Capability(
+                    PROXY_HOSTS,
+                    CapabilityOperation.VERIFY,
+                ),
+            }
+        )
+
+    def connect(self):
+        return None
+
+    def discover(self, object_type):
+        if object_type == CERTIFICATES:
+            return [
+                DiscoveredObject(
+                    object_type=CERTIFICATES,
+                    local_key="67",
+                    payload={
+                        "domain_names": [
+                            "app.example.com",
+                        ],
+                        "provider": "letsencrypt",
+                        "nice_name": "app",
+                        "valid_from": (
+                            "2026-09-01T00:00:00Z"
+                        ),
+                        "expires_on": (
+                            "2026-12-01T00:00:00Z"
+                        ),
+                        "cluster_managed": True,
+                        "acme_renew": True,
+                        "dns_challenge": False,
+                    },
+                ),
+            ]
+
+        if object_type == PROXY_HOSTS:
+            return [
+                DiscoveredObject(
+                    object_type=PROXY_HOSTS,
+                    local_key="12",
+                    payload={
+                        "domain_names": [
+                            "app.example.com",
+                        ],
+                        "forward_scheme": "http",
+                        "forward_host": "backend.internal",
+                        "forward_port": 8080,
+                        "ssl_forced": True,
+                        "caching_enabled": False,
+                        "block_exploits": True,
+                        "allow_websocket_upgrade": True,
+                        "http2_support": True,
+                        "hsts_enabled": True,
+                        "hsts_subdomains": False,
+                        "enabled": True,
+                    },
+                    references=(
+                        ObjectReference(
+                            relation="certificate",
+                            object_type=CERTIFICATES,
+                            local_key="67",
+                        ),
+                    ),
+                ),
+            ]
+
+        raise RuntimeError(
+            f"unsupported discovery type: {object_type}"
+        )
+
+    def apply(self, action):
+        raise AssertionError(
+            "NPM APPLY must remain unavailable"
+        )
+
+    def verify(self, object):
+        return True
+
+
+class NpmDesiredStateOrchestrationTests(
+    unittest.TestCase
+):
+    def test_orchestrator_builds_and_persists_all_three_desired_types(
+        self,
+    ):
+        adapter = FakeNpmAdapter(
+            Endpoint(
+                "npm",
+                "192.168.1.3:81",
+            )
+        )
+
+        with SQLiteStateStore(":memory:") as store:
+            run = discover_reconcile_plan(
+                [adapter],
+                store,
+                target_ip="192.168.1.101",
+                run_at=NOW,
+            )
+
+            desired_by_type = {
+                object.object_type: object
+                for object in run.desired
+            }
+
+            self.assertEqual(
+                set(desired_by_type),
+                {
+                    CERTIFICATES,
+                    PROXY_HOSTS,
+                    DNS_RECORDS,
+                },
+            )
+
+            certificate = desired_by_type[
+                CERTIFICATES
+            ]
+            proxy = desired_by_type[PROXY_HOSTS]
+            dns = desired_by_type[DNS_RECORDS]
+
+            self.assertIsNone(
+                certificate.local_key
+            )
+            self.assertIsNone(
+                proxy.local_key
+            )
+            self.assertIsNone(
+                dns.local_key
+            )
+
+            self.assertEqual(
+                proxy.references,
+                (
+                    ObjectReference(
+                        relation="certificate",
+                        object_type=CERTIFICATES,
+                        logical_key="app.example.com",
+                    ),
+                ),
+            )
+
+            self.assertEqual(
+                dns.payload["hostname"],
+                "app.example.com",
+            )
+            self.assertEqual(
+                dns.payload["ip"],
+                "192.168.1.101",
+            )
+
+            for object_type in (
+                CERTIFICATES,
+                PROXY_HOSTS,
+                DNS_RECORDS,
+            ):
+                persisted = (
+                    store.load_desired_state(
+                        object_type
+                    )
+                )
+
+                self.assertEqual(
+                    len(persisted),
+                    1,
+                )
+                self.assertEqual(
+                    persisted[0].object,
+                    desired_by_type[
+                        object_type
+                    ],
+                )
+
+            # The NPM adapter intentionally does not advertise
+            # APPLY, so this read-only orchestration must not
+            # produce an NPM endpoint plan.
+            self.assertEqual(
+                run.plans,
+                (),
+            )
+
+    def test_empty_npm_snapshot_clears_stale_npm_desired_state(
+        self,
+    ):
+        adapter = FakeNpmAdapter(
+            Endpoint(
+                "npm",
+                "192.168.1.3:81",
+            )
+        )
+
+        with SQLiteStateStore(":memory:") as store:
+            first = discover_reconcile_plan(
+                [adapter],
+                store,
+                target_ip="192.168.1.101",
+                run_at=NOW,
+            )
+
+            self.assertTrue(first.desired)
+            self.assertEqual(
+                len(
+                    store.load_desired_state(
+                        CERTIFICATES
+                    )
+                ),
+                1,
+            )
+            self.assertEqual(
+                len(
+                    store.load_desired_state(
+                        PROXY_HOSTS
+                    )
+                ),
+                1,
+            )
+
+            class EmptyNpmAdapter(
+                FakeNpmAdapter
+            ):
+                def discover(
+                    self,
+                    object_type,
+                ):
+                    if object_type in {
+                        CERTIFICATES,
+                        PROXY_HOSTS,
+                    }:
+                        return []
+
+                    return super().discover(
+                        object_type
+                    )
+
+            empty = EmptyNpmAdapter(
+                adapter.endpoint
+            )
+
+            second = discover_reconcile_plan(
+                [empty],
+                store,
+                target_ip="192.168.1.101",
+                run_at=NOW.replace(minute=31),
+            )
+
+            self.assertEqual(
+                second.desired,
+                (),
+            )
+            self.assertEqual(
+                store.load_desired_state(
+                    CERTIFICATES
+                ),
+                [],
+            )
+            self.assertEqual(
+                store.load_desired_state(
+                    PROXY_HOSTS
+                ),
+                [],
+            )
+            self.assertEqual(
+                store.load_desired_state(
+                    DNS_RECORDS
+                ),
+                [],
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_planning.py

@@ -289,6 +289,543 @@ class EndpointPlanningTests(unittest.TestCase):
             )
 
 
+    def test_certificate_provider_difference_is_noop_when_portable_state_matches(self):
+        endpoint = Endpoint("npm", "node-two:81")
+
+        npm_adapter = FakeAdapter(
+            endpoint=endpoint,
+            capabilities=frozenset(
+                {
+                    Capability(
+                        CERTIFICATES,
+                        CapabilityOperation.APPLY,
+                    ),
+                }
+            ),
+        )
+
+        desired = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "provider": "letsencrypt",
+                "nice_name": "source",
+                "valid_from": "2026-09-01T00:00:00+00:00",
+                "expires_on": "2026-12-01T00:00:00+00:00",
+            },
+        )
+
+        current = Observation(
+            endpoint_id=endpoint.identity,
+            object=NormalizedObject(
+                object_type=CERTIFICATES,
+                logical_key="app.example.com",
+                local_key="84",
+                payload={
+                    "domain_names": ["app.example.com"],
+                    "provider": "other",
+                    "nice_name": "imported copy",
+                    "valid_from": "2026-09-01T00:00:00+00:00",
+                    "expires_on": "2026-12-01T00:00:00+00:00",
+                },
+            ),
+            observed_at=NOW,
+        )
+
+        plans = build_endpoint_plans(
+            [npm_adapter],
+            [desired],
+            [current],
+        )
+
+        self.assertEqual(
+            plans[0].actions[0].operation,
+            PlanOperation.NOOP,
+        )
+
+    def test_proxy_reference_change_is_update_even_when_payload_matches(self):
+        endpoint = Endpoint("npm", "node-two:81")
+
+        npm_adapter = FakeAdapter(
+            endpoint=endpoint,
+            capabilities=frozenset(
+                {
+                    Capability(
+                        PROXY_HOSTS,
+                        CapabilityOperation.APPLY,
+                    ),
+                }
+            ),
+        )
+
+        desired = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "forward_scheme": "http",
+                "forward_host": "backend.internal",
+                "forward_port": 8080,
+            },
+            references=(
+                ObjectReference(
+                    relation="certificate",
+                    object_type=CERTIFICATES,
+                    logical_key="new.example.com",
+                ),
+            ),
+        )
+
+        observations = [
+            Observation(
+                endpoint_id=endpoint.identity,
+                object=NormalizedObject(
+                    object_type=CERTIFICATES,
+                    logical_key="old.example.com",
+                    local_key="67",
+                    payload={
+                        "domain_names": ["old.example.com"],
+                    },
+                ),
+                observed_at=NOW,
+            ),
+            Observation(
+                endpoint_id=endpoint.identity,
+                object=NormalizedObject(
+                    object_type=PROXY_HOSTS,
+                    logical_key="app.example.com",
+                    local_key="12",
+                    payload=dict(desired.payload),
+                    references=(
+                        ObjectReference(
+                            relation="certificate",
+                            object_type=CERTIFICATES,
+                            local_key="67",
+                        ),
+                    ),
+                ),
+                observed_at=NOW,
+            ),
+        ]
+
+        plans = build_endpoint_plans(
+            [npm_adapter],
+            [desired],
+            observations,
+        )
+
+        self.assertEqual(
+            plans[0].actions[0].operation,
+            PlanOperation.UPDATE,
+        )
+
+    def test_planner_logicalizes_matching_local_reference_to_noop(self):
+        endpoint = Endpoint("npm", "node-two:81")
+
+        npm_adapter = FakeAdapter(
+            endpoint=endpoint,
+            capabilities=frozenset(
+                {
+                    Capability(
+                        PROXY_HOSTS,
+                        CapabilityOperation.APPLY,
+                    ),
+                }
+            ),
+        )
+
+        desired = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "forward_scheme": "http",
+                "forward_host": "backend.internal",
+                "forward_port": 8080,
+            },
+            references=(
+                ObjectReference(
+                    relation="certificate",
+                    object_type=CERTIFICATES,
+                    logical_key="app.example.com",
+                ),
+            ),
+        )
+
+        observations = [
+            Observation(
+                endpoint_id=endpoint.identity,
+                object=NormalizedObject(
+                    object_type=CERTIFICATES,
+                    logical_key="app.example.com",
+                    local_key="84",
+                    payload={
+                        "domain_names": ["app.example.com"],
+                    },
+                ),
+                observed_at=NOW,
+            ),
+            Observation(
+                endpoint_id=endpoint.identity,
+                object=NormalizedObject(
+                    object_type=PROXY_HOSTS,
+                    logical_key="app.example.com",
+                    local_key="12",
+                    payload=dict(desired.payload),
+                    references=(
+                        ObjectReference(
+                            relation="certificate",
+                            object_type=CERTIFICATES,
+                            local_key="84",
+                        ),
+                    ),
+                ),
+                observed_at=NOW,
+            ),
+        ]
+
+        plans = build_endpoint_plans(
+            [npm_adapter],
+            [desired],
+            observations,
+        )
+
+        self.assertEqual(
+            plans[0].actions[0].operation,
+            PlanOperation.NOOP,
+        )
+        self.assertEqual(
+            plans[0].actions[0].current.references,
+            desired.references,
+        )
+
+    def test_older_certificate_is_replace_and_dependent_proxy_is_update(self):
+        endpoint = Endpoint(
+            "generic",
+            "node.example:443",
+        )
+
+        generic_adapter = adapter(
+            "generic",
+            "node.example:443",
+            CERTIFICATES,
+            PROXY_HOSTS,
+        )
+
+        desired_certificate = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "valid_from": "2026-09-20T00:00:00+00:00",
+                "expires_on": "2026-12-20T00:00:00+00:00",
+            },
+        )
+
+        desired_proxy = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "forward_scheme": "https",
+                "forward_host": "backend.internal",
+                "forward_port": 443,
+                "enabled": True,
+            },
+            references=(
+                ObjectReference(
+                    relation="certificate",
+                    object_type=CERTIFICATES,
+                    logical_key="app.example.com",
+                ),
+            ),
+        )
+
+        current_certificate = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            local_key="67",
+            payload={
+                "domain_names": ["app.example.com"],
+                "valid_from": "2026-08-20T00:00:00+00:00",
+                "expires_on": "2026-11-20T00:00:00+00:00",
+            },
+        )
+
+        current_proxy = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            local_key="12",
+            payload=dict(desired_proxy.payload),
+            references=(
+                ObjectReference(
+                    relation="certificate",
+                    object_type=CERTIFICATES,
+                    local_key="67",
+                ),
+            ),
+        )
+
+        plans = build_endpoint_plans(
+            [generic_adapter],
+            [
+                desired_certificate,
+                desired_proxy,
+            ],
+            [
+                observed(
+                    endpoint.identity,
+                    current_certificate,
+                ),
+                observed(
+                    endpoint.identity,
+                    current_proxy,
+                ),
+            ],
+        )
+
+        self.assertEqual(len(plans), 1)
+
+        actions = plans[0].actions
+
+        self.assertEqual(
+            [
+                (
+                    action.operation,
+                    action.desired.object_type,
+                )
+                for action in actions
+            ],
+            [
+                (
+                    PlanOperation.REPLACE,
+                    CERTIFICATES,
+                ),
+                (
+                    PlanOperation.UPDATE,
+                    PROXY_HOSTS,
+                ),
+            ],
+        )
+
+        self.assertEqual(
+            actions[1].reason,
+            "referenced object is being replaced",
+        )
+
+
+    def test_restart_duplicate_certificate_uses_desired_copy_and_rebinds_proxy(self):
+        endpoint = Endpoint(
+            "generic",
+            "node.example:443",
+        )
+
+        generic_adapter = adapter(
+            "generic",
+            "node.example:443",
+            CERTIFICATES,
+            PROXY_HOSTS,
+        )
+
+        desired_certificate = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "provider": "letsencrypt",
+                "valid_from": "2026-09-20T00:00:00+00:00",
+                "expires_on": "2026-12-20T00:00:00+00:00",
+            },
+        )
+
+        desired_proxy = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "forward_scheme": "https",
+                "forward_host": "backend.internal",
+                "forward_port": 443,
+                "enabled": True,
+            },
+            references=(
+                ObjectReference(
+                    relation="certificate",
+                    object_type=CERTIFICATES,
+                    logical_key="app.example.com",
+                ),
+            ),
+        )
+
+        old_certificate = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            local_key="67",
+            payload={
+                "domain_names": ["app.example.com"],
+                "provider": "other",
+                "valid_from": "2026-08-20T00:00:00+00:00",
+                "expires_on": "2026-11-20T00:00:00+00:00",
+            },
+        )
+
+        new_certificate = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            local_key="84",
+            payload={
+                "domain_names": ["app.example.com"],
+                "provider": "other",
+                "valid_from": "2026-09-20T00:00:00+00:00",
+                "expires_on": "2026-12-20T00:00:00+00:00",
+            },
+        )
+
+        current_proxy = NormalizedObject(
+            object_type=PROXY_HOSTS,
+            logical_key="app.example.com",
+            local_key="12",
+            payload=dict(desired_proxy.payload),
+            references=(
+                ObjectReference(
+                    relation="certificate",
+                    object_type=CERTIFICATES,
+                    local_key="67",
+                ),
+            ),
+        )
+
+        plans = build_endpoint_plans(
+            [generic_adapter],
+            [
+                desired_certificate,
+                desired_proxy,
+            ],
+            [
+                observed(
+                    endpoint.identity,
+                    old_certificate,
+                ),
+                observed(
+                    endpoint.identity,
+                    new_certificate,
+                ),
+                observed(
+                    endpoint.identity,
+                    current_proxy,
+                ),
+            ],
+        )
+
+        self.assertEqual(len(plans), 1)
+
+        plan = plans[0]
+
+        by_type = {
+            action.desired.object_type: action
+            for action in plan.actions
+        }
+
+        self.assertEqual(
+            by_type[CERTIFICATES].operation,
+            PlanOperation.NOOP,
+        )
+        self.assertEqual(
+            by_type[CERTIFICATES].current.local_key,
+            "84",
+        )
+
+        self.assertEqual(
+            by_type[PROXY_HOSTS].operation,
+            PlanOperation.UPDATE,
+        )
+        self.assertEqual(
+            by_type[PROXY_HOSTS].reason,
+            "referenced local object is being retired",
+        )
+
+        self.assertEqual(
+            tuple(
+                retirement.object.local_key
+                for retirement in plan.retirements
+            ),
+            ("67",),
+        )
+
+    def test_equivalent_duplicate_preserves_letsencrypt_copy(self):
+        endpoint = Endpoint(
+            "generic",
+            "node.example:443",
+        )
+
+        generic_adapter = adapter(
+            "generic",
+            "node.example:443",
+            CERTIFICATES,
+        )
+
+        desired = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "valid_from": "2026-09-20T00:00:00+00:00",
+                "expires_on": "2026-12-20T00:00:00+00:00",
+            },
+        )
+
+        imported = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            local_key="90",
+            payload={
+                "domain_names": ["app.example.com"],
+                "provider": "other",
+                "valid_from": "2026-09-20T00:00:00+00:00",
+                "expires_on": "2026-12-20T00:00:00+00:00",
+            },
+        )
+
+        native = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            local_key="12",
+            payload={
+                "domain_names": ["app.example.com"],
+                "provider": "letsencrypt",
+                "valid_from": "2026-09-20T00:00:00+00:00",
+                "expires_on": "2026-12-20T00:00:00+00:00",
+            },
+        )
+
+        plans = build_endpoint_plans(
+            [generic_adapter],
+            [desired],
+            [
+                observed(endpoint.identity, imported),
+                observed(endpoint.identity, native),
+            ],
+        )
+
+        action = plans[0].actions[0]
+
+        self.assertEqual(
+            action.operation,
+            PlanOperation.NOOP,
+        )
+        self.assertEqual(
+            action.current.local_key,
+            "12",
+        )
+        self.assertEqual(
+            tuple(
+                retirement.object.local_key
+                for retirement in plans[0].retirements
+            ),
+            ("90",),
+        )
+
     def test_duplicate_endpoint_objects_fail_without_payload_output(self):
         desired = [
             dns_object("app.example.com", "192.168.1.101")

tests/test_reconciliation.py

@@ -4,12 +4,15 @@ from datetime import datetime, timezone
 from dns_sync.adapters import (
     CERTIFICATES,
     DNS_RECORDS,
+    PROXY_HOSTS,
     NormalizedObject,
+    ObjectReference,
     Observation,
 )
 from dns_sync.reconciliation import (
     ReconciliationConflict,
     group_observations,
+    logicalize_observation_references,
     reconcile_observations,
 )
 
@@ -166,12 +169,12 @@ class DnsUnionPolicyTests(unittest.TestCase):
 
     def test_missing_policy_fails_explicitly(self):
         observation = Observation(
-            endpoint_id="npm://node-one:81",
+            endpoint_id="generic://node-one",
             object=NormalizedObject(
-                object_type=CERTIFICATES,
+                object_type="unknown_objects",
                 logical_key="app.example.com",
                 local_key="67",
-                payload={"expires_on": "2026-12-15"},
+                payload={},
             ),
             observed_at=NOW,
         )
@@ -182,6 +185,304 @@ class DnsUnionPolicyTests(unittest.TestCase):
         ):
             reconcile_observations([observation])
 
+    def test_local_certificate_references_become_logical_references(self):
+        observations = [
+            Observation(
+                endpoint_id="npm://node-one:81",
+                object=NormalizedObject(
+                    object_type=CERTIFICATES,
+                    logical_key="app.example.com",
+                    local_key="67",
+                    payload={
+                        "domain_names": ["app.example.com"],
+                        "valid_from": "2026-09-01T00:00:00+00:00",
+                        "expires_on": "2026-12-01T00:00:00+00:00",
+                    },
+                ),
+                observed_at=NOW,
+            ),
+            Observation(
+                endpoint_id="npm://node-one:81",
+                object=NormalizedObject(
+                    object_type=PROXY_HOSTS,
+                    logical_key="app.example.com",
+                    local_key="12",
+                    payload={
+                        "domain_names": ["app.example.com"],
+                        "forward_scheme": "http",
+                        "forward_host": "backend.internal",
+                        "forward_port": 8080,
+                        "enabled": True,
+                    },
+                    references=(
+                        ObjectReference(
+                            relation="certificate",
+                            object_type=CERTIFICATES,
+                            local_key="67",
+                        ),
+                    ),
+                ),
+                observed_at=NOW,
+            ),
+        ]
+
+        logicalized = logicalize_observation_references(
+            observations
+        )
+
+        proxy = next(
+            row.object
+            for row in logicalized
+            if row.object.object_type == PROXY_HOSTS
+        )
+
+        self.assertEqual(
+            proxy.references,
+            (
+                ObjectReference(
+                    relation="certificate",
+                    object_type=CERTIFICATES,
+                    logical_key="app.example.com",
+                ),
+            ),
+        )
+
+    def test_unresolved_local_reference_fails_closed(self):
+        observation = Observation(
+            endpoint_id="npm://node-one:81",
+            object=NormalizedObject(
+                object_type=PROXY_HOSTS,
+                logical_key="app.example.com",
+                local_key="12",
+                payload={
+                    "domain_names": ["app.example.com"],
+                    "forward_scheme": "http",
+                    "forward_host": "backend.internal",
+                    "forward_port": 8080,
+                },
+                references=(
+                    ObjectReference(
+                        relation="certificate",
+                        object_type=CERTIFICATES,
+                        local_key="999",
+                    ),
+                ),
+            ),
+            observed_at=NOW,
+        )
+
+        with self.assertRaisesRegex(
+            ReconciliationConflict,
+            "Unresolved endpoint-local reference",
+        ):
+            reconcile_observations([observation])
+
+    def test_certificate_policy_selects_newest_expiry(self):
+        observations = [
+            Observation(
+                endpoint_id="npm://node-one:81",
+                object=NormalizedObject(
+                    object_type=CERTIFICATES,
+                    logical_key="app.example.com",
+                    local_key="10",
+                    payload={
+                        "domain_names": ["app.example.com"],
+                        "provider": "other",
+                        "nice_name": "older",
+                        "valid_from": "2026-06-01T00:00:00+00:00",
+                        "expires_on": "2026-09-01T00:00:00+00:00",
+                    },
+                ),
+                observed_at=NOW,
+            ),
+            Observation(
+                endpoint_id="npm://node-two:81",
+                object=NormalizedObject(
+                    object_type=CERTIFICATES,
+                    logical_key="app.example.com",
+                    local_key="44",
+                    payload={
+                        "domain_names": ["app.example.com"],
+                        "provider": "letsencrypt",
+                        "nice_name": "newer",
+                        "valid_from": "2026-09-01T00:00:00+00:00",
+                        "expires_on": "2026-12-01T00:00:00+00:00",
+                    },
+                ),
+                observed_at=NOW,
+            ),
+        ]
+
+        desired = reconcile_observations(observations)
+
+        self.assertEqual(len(desired), 1)
+        self.assertEqual(
+            desired[0].payload["expires_on"],
+            "2026-12-01T00:00:00+00:00",
+        )
+        self.assertEqual(
+            desired[0].payload["provider"],
+            "letsencrypt",
+        )
+        self.assertIsNone(desired[0].local_key)
+
+    def test_certificate_policy_prefers_newer_valid_from_over_expiry(self):
+        observations = [
+            Observation(
+                endpoint_id="npm://node-one:81",
+                object=NormalizedObject(
+                    object_type=CERTIFICATES,
+                    logical_key="app.example.com",
+                    local_key="10",
+                    payload={
+                        "domain_names": ["app.example.com"],
+                        "valid_from": "2026-08-01T00:00:00+00:00",
+                        "expires_on": "2027-01-01T00:00:00+00:00",
+                    },
+                ),
+                observed_at=NOW,
+            ),
+            Observation(
+                endpoint_id="npm://node-two:81",
+                object=NormalizedObject(
+                    object_type=CERTIFICATES,
+                    logical_key="app.example.com",
+                    local_key="44",
+                    payload={
+                        "domain_names": ["app.example.com"],
+                        "valid_from": "2026-09-15T00:00:00+00:00",
+                        "expires_on": "2026-12-15T00:00:00+00:00",
+                    },
+                ),
+                observed_at=NOW,
+            ),
+        ]
+
+        desired = reconcile_observations(observations)
+
+        self.assertEqual(len(desired), 1)
+        self.assertEqual(
+            desired[0].payload["valid_from"],
+            "2026-09-15T00:00:00+00:00",
+        )
+        self.assertEqual(
+            desired[0].payload["expires_on"],
+            "2026-12-15T00:00:00+00:00",
+        )
+
+    def test_proxy_consensus_ignores_different_endpoint_local_certificate_ids(self):
+        observations = []
+
+        for endpoint_id, certificate_id, proxy_id in (
+            ("npm://node-one:81", "67", "12"),
+            ("npm://node-two:81", "84", "31"),
+        ):
+            observations.extend(
+                [
+                    Observation(
+                        endpoint_id=endpoint_id,
+                        object=NormalizedObject(
+                            object_type=CERTIFICATES,
+                            logical_key="app.example.com",
+                            local_key=certificate_id,
+                            payload={
+                                "domain_names": ["app.example.com"],
+                                "valid_from": "2026-09-01T00:00:00+00:00",
+                                "expires_on": "2026-12-01T00:00:00+00:00",
+                            },
+                        ),
+                        observed_at=NOW,
+                    ),
+                    Observation(
+                        endpoint_id=endpoint_id,
+                        object=NormalizedObject(
+                            object_type=PROXY_HOSTS,
+                            logical_key="app.example.com",
+                            local_key=proxy_id,
+                            payload={
+                                "domain_names": ["app.example.com"],
+                                "forward_scheme": "http",
+                                "forward_host": "backend.internal",
+                                "forward_port": 8080,
+                                "enabled": True,
+                            },
+                            references=(
+                                ObjectReference(
+                                    relation="certificate",
+                                    object_type=CERTIFICATES,
+                                    local_key=certificate_id,
+                                ),
+                            ),
+                        ),
+                        observed_at=NOW,
+                    ),
+                ]
+            )
+
+        desired = reconcile_observations(observations)
+
+        proxy = next(
+            object
+            for object in desired
+            if object.object_type == PROXY_HOSTS
+        )
+
+        self.assertEqual(
+            proxy.references,
+            (
+                ObjectReference(
+                    relation="certificate",
+                    object_type=CERTIFICATES,
+                    logical_key="app.example.com",
+                ),
+            ),
+        )
+
+    def test_proxy_consensus_rejects_backend_drift_without_payload_output(self):
+        observations = [
+            Observation(
+                endpoint_id="npm://node-one:81",
+                object=NormalizedObject(
+                    object_type=PROXY_HOSTS,
+                    logical_key="app.example.com",
+                    local_key="12",
+                    payload={
+                        "domain_names": ["app.example.com"],
+                        "forward_scheme": "http",
+                        "forward_host": "one.internal",
+                        "forward_port": 8080,
+                    },
+                ),
+                observed_at=NOW,
+            ),
+            Observation(
+                endpoint_id="npm://node-two:81",
+                object=NormalizedObject(
+                    object_type=PROXY_HOSTS,
+                    logical_key="app.example.com",
+                    local_key="31",
+                    payload={
+                        "domain_names": ["app.example.com"],
+                        "forward_scheme": "http",
+                        "forward_host": "two.internal",
+                        "forward_port": 8080,
+                    },
+                ),
+                observed_at=NOW,
+            ),
+        ]
+
+        with self.assertRaises(ReconciliationConflict) as raised:
+            reconcile_observations(observations)
+
+        message = str(raised.exception)
+
+        self.assertIn("app.example.com", message)
+        self.assertIn("npm://node-one:81", message)
+        self.assertIn("npm://node-two:81", message)
+        self.assertNotIn("one.internal", message)
+        self.assertNotIn("two.internal", message)
+
     def test_explicit_empty_policy_registry_uses_no_defaults(self):
         observation = dns_observation(
             "npm://node-one:81",