fix(runtime): refuse successful backend claims that diverge from the plan

implementations/python/packages/raes_runtime/backend_calls.py

@@ -22,10 +22,12 @@
     sanitize_account_credential_result,
     value_free_backend_diagnostics,
 )
+from .backend_plan_conformance import plan_conformance_diagnostics
 from .backend_realization_authority import (
     _apply_authority_diagnostics,
     _bind_submitted_plan,
     _RealizationApplyContext,
+    _supplemental_realization_requirements,
 )
 from .diagnostics import _failure_diagnostic
 from .evaluation_result_contracts import evaluation_result_contract_diagnostics
@@ -183,6 +185,8 @@ def _post_apply_contract_result(
         baseline_snapshot,
         information_state_context_resolver=information_state_context_resolver,
     )
+    if not diagnostics and realization.plan is not None:
+        diagnostics = plan_conformance_diagnostics(result, baseline_snapshot, realization.plan)
     provenance: tuple[RealizationProvenanceEntry, ...] = ()
     if not diagnostics and realization.plan is not None:
         diagnostics, provenance = realization_authority_disclosure(
@@ -263,23 +267,6 @@ def _sanitize_backend_realization(
     return sanitized
 
 
-def _supplemental_realization_requirements(
-    realization: _RealizationApplyContext,
-) -> tuple[CompiledRealizationRequirement, ...]:
-    """Keep non-registry contracts while the plan owns registry concerns."""
-
-    if realization.plan is None or not realization.plan.realization_authority:
-        return realization.requirements
-    plan_identities = {
-        (entry.address, entry.field_path, entry.requirement_kind) for entry in realization.plan.realization_authority
-    }
-    return tuple(
-        requirement
-        for requirement in realization.requirements
-        if (requirement.address, requirement.field_path, requirement.requirement_kind) not in plan_identities
-    )
-
-
 def _with_snapshot(
     result: ApplyResult,
     snapshot: RuntimeSnapshot,

implementations/python/packages/raes_runtime/backend_plan_conformance.py

@@ -0,0 +1,98 @@
+"""Plan-conformance gate for a provisioning backend's realized snapshot.
+
+The snapshot-contract checks in :mod:`raes_runtime.backend_calls` validate a
+returned snapshot against its own shape and the baseline it evolved from; this
+module closes the issue-158 falsification matrix by additionally holding the
+result to the *submitted plan*. A backend claim is refused when it realizes an
+address the plan never authorized, relabels a planned resource's type, files a
+planned resource under a different domain, or mutates state it does not
+disclose in ``changed_addresses``.
+"""
+
+from __future__ import annotations
+
+from raes_contracts.diagnostics import Diagnostic, Severity
+from raes_contracts.planning import ChangeAction, ProvisioningPlan, RuntimeDomain
+from raes_contracts.runtime_state import ApplyResult, RuntimeSnapshot
+
+_BACKEND_CONTRACT_INVALID = "runtime.backend-contract-invalid"
+
+
+def _conformance_diagnostic(address: str, message: str) -> Diagnostic:
+    return Diagnostic(
+        code=_BACKEND_CONTRACT_INVALID,
+        domain="runtime",
+        address=address,
+        message=message,
+        severity=Severity.ERROR,
+    )
+
+
+def _entry_conformance_message(
+    entry: object,
+    operation: object,
+    expected_domain: RuntimeDomain,
+) -> str | None:
+    if entry.resource_type != operation.resource_type:
+        return (
+            f"Backend misreported the resource type at '{entry.address}': the plan requires "
+            f"'{operation.resource_type}' but the backend claims '{entry.resource_type}'."
+        )
+    if entry.domain is not expected_domain:
+        claimed = getattr(entry.domain, "value", entry.domain)
+        return (
+            f"Backend filed '{entry.address}' under domain '{claimed}'; the provisioning plan "
+            f"authorizes '{expected_domain.value}'."
+        )
+    return None
+
+
+def _expected_domain(plan: ProvisioningPlan, address: str) -> RuntimeDomain:
+    planned_resource = plan.resources.get(address)
+    return planned_resource.domain if planned_resource is not None else RuntimeDomain.PROVISIONING
+
+
+def plan_conformance_diagnostics(
+    result: ApplyResult,
+    baseline_snapshot: RuntimeSnapshot,
+    plan: ProvisioningPlan,
+) -> list[Diagnostic]:
+    """Refuse realized entries the submitted provisioning plan never authorized.
+
+    Applies to successful claims only: a failed apply already returns a
+    non-authoritative result, and its partial snapshot is preserved for
+    forensics rather than being replaced by the baseline.
+    """
+
+    if not result.success:
+        return []
+    operations = {operation.address: operation for operation in plan.operations}
+    diagnostics: list[Diagnostic] = []
+    disclosed = set(result.changed_addresses)
+    for address, entry in result.snapshot.entries.items():
+        baseline_entry = baseline_snapshot.entries.get(address)
+        operation = operations.get(address)
+        changed = baseline_entry is None or entry != baseline_entry
+        if operation is None:
+            if changed:
+                verb = "realized" if baseline_entry is None else "mutated"
+                diagnostics.append(
+                    _conformance_diagnostic(
+                        address,
+                        f"Backend {verb} '{address}', an address absent from the provisioning plan.",
+                    )
+                )
+            continue
+        if operation.action is ChangeAction.DELETE:
+            continue
+        message = _entry_conformance_message(entry, operation, _expected_domain(plan, address))
+        if message is not None:
+            diagnostics.append(_conformance_diagnostic(address, message))
+        elif changed and address not in disclosed and operation.action is not ChangeAction.UNCHANGED:
+            diagnostics.append(
+                _conformance_diagnostic(
+                    address,
+                    f"Backend changed '{address}' without disclosing it in changed_addresses.",
+                )
+            )
+    return diagnostics

implementations/python/packages/raes_runtime/backend_realization_authority.py

@@ -70,3 +70,20 @@ def _bind_submitted_plan(
 
 
 __all__ = ["_RealizationApplyContext", "_apply_authority_diagnostics", "_bind_submitted_plan"]
+
+
+def _supplemental_realization_requirements(
+    realization: _RealizationApplyContext,
+) -> tuple[CompiledRealizationRequirement, ...]:
+    """Keep non-registry contracts while the plan owns registry concerns."""
+
+    if realization.plan is None or not realization.plan.realization_authority:
+        return realization.requirements
+    plan_identities = {
+        (entry.address, entry.field_path, entry.requirement_kind) for entry in realization.plan.realization_authority
+    }
+    return tuple(
+        requirement
+        for requirement in realization.requirements
+        if (requirement.address, requirement.field_path, requirement.requirement_kind) not in plan_identities
+    )

implementations/python/tests/test_issue_158_backend_falsification_matrix.py

@@ -0,0 +1,189 @@
+"""Does the backend boundary falsify dishonest realization claims? (issue #158)
+
+Issue #158 is a falsification protocol, not an implementation design. Its claim
+under test: *backend agnosticism is credible only if conformance can reject
+dishonest, incomplete, or overbroad backend claims instead of merely validating
+happy-path stubs.*
+
+Method. Each case is the **real** `ReferenceProvisioner` perturbed by exactly one
+lie, so a rejection is attributable to that lie rather than to an incomplete
+hand-rolled double. `test_honest_baseline_is_accepted` is the control: without a
+perturbation the same construction is admitted, which is what makes the other
+results meaningful.
+
+Every fabrication below is refused at the boundary: the runtime returns the
+untouched baseline snapshot with ``runtime.backend-contract-invalid``, so no lie
+survives into the authoritative snapshot. "Realizes nothing" is refused by the
+SEM-218 realization gate; the other four are refused by the plan-conformance
+gate (``raes_runtime.backend_plan_conformance``), which holds the returned
+snapshot to the submitted plan's addresses, resource types, domains, and
+disclosed changes.
+
+These probes bound the runtime backend boundary (`RuntimeManager.apply`) only.
+`run_target_conformance` is not used as the discriminator because with default
+arguments it also fails the honest reference backend, for the reason recorded in
+issue #663: its default scenario is not universally realizable.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+
+from raes import parse_sdl
+from raes_contracts.planning import RuntimeDomain
+from raes_contracts.runtime_state import ApplyResult
+from raes_reference_backend import create_reference_backend_target
+from raes_reference_backend.provisioner import ReferenceProvisioner
+from raes_runtime.manager import RuntimeManager
+
+_SCENARIO = """
+name: backend-falsification
+nodes:
+  lab: {type: switch}
+  web: {type: compute, resources: {ram: 1 gib, cpu: 1}}
+infrastructure:
+  lab: {count: 1, properties: {cidr: 10.0.0.0/24, gateway: 10.0.0.1}}
+  web: {count: 1, links: [lab]}
+"""
+
+_SMUGGLED_ADDRESS = "provision.node.smuggled"
+
+
+class _RealizesNothing(ReferenceProvisioner):
+    """Reports success and every address as changed, returning the predecessor."""
+
+    def apply(self, plan, snapshot):
+        super().apply(plan, snapshot)
+        return ApplyResult(
+            success=True,
+            snapshot=snapshot,
+            changed_addresses=[operation.address for operation in plan.operations],
+        )
+
+
+class _InventsResource(ReferenceProvisioner):
+    """Realizes the plan, then adds a resource the scenario never authored."""
+
+    def apply(self, plan, snapshot):
+        result = super().apply(plan, snapshot)
+        if not result.success:
+            return result
+        entries = dict(result.snapshot.entries)
+        entries[_SMUGGLED_ADDRESS] = dataclasses.replace(next(iter(entries.values())), address=_SMUGGLED_ADDRESS)
+        return dataclasses.replace(
+            result,
+            snapshot=result.snapshot.with_entries(entries),
+            changed_addresses=[*result.changed_addresses, _SMUGGLED_ADDRESS],
+        )
+
+
+class _SubstitutesResourceType(ReferenceProvisioner):
+    """Realizes the planned addresses but relabels a node as a network."""
+
+    def apply(self, plan, snapshot):
+        result = super().apply(plan, snapshot)
+        if not result.success:
+            return result
+        entries = dict(result.snapshot.entries)
+        for address, entry in entries.items():
+            if entry.resource_type == "node":
+                entries[address] = dataclasses.replace(entry, resource_type="network")
+                break
+        return dataclasses.replace(result, snapshot=result.snapshot.with_entries(entries))
+
+
+class _UnderReportsChanges(ReferenceProvisioner):
+    """Mutates the snapshot but reports no changed addresses."""
+
+    def apply(self, plan, snapshot):
+        result = super().apply(plan, snapshot)
+        if not result.success:
+            return result
+        return dataclasses.replace(result, changed_addresses=[])
+
+
+class _ForgesDomain(ReferenceProvisioner):
+    """Files provisioning results under the evaluation domain."""
+
+    def apply(self, plan, snapshot):
+        result = super().apply(plan, snapshot)
+        if not result.success:
+            return result
+        entries = {
+            address: dataclasses.replace(entry, domain=RuntimeDomain.EVALUATION)
+            for address, entry in result.snapshot.entries.items()
+        }
+        return dataclasses.replace(result, snapshot=result.snapshot.with_entries(entries))
+
+
+def _apply_with(provisioner_class=None):
+    """Apply the scenario through the reference target, optionally perturbed.
+
+    The perturbed provisioner is built from the honest one's driver and
+    realization envelope, so the only difference from the control is the lie.
+    """
+
+    target = create_reference_backend_target()
+    if provisioner_class is not None:
+        honest = target.provisioner
+        target = dataclasses.replace(
+            target,
+            provisioner=provisioner_class(
+                honest._driver,  # noqa: SLF001 - perturbing the real provisioner is the method
+                realization_envelope=honest._realization_envelope,  # noqa: SLF001
+            ),
+        )
+    manager = RuntimeManager(target)
+    return manager.apply(manager.plan(parse_sdl(_SCENARIO)))
+
+
+def test_honest_baseline_is_accepted():
+    """The control. Without it, a rejection below would prove nothing."""
+
+    result = _apply_with()
+
+    assert result.success, [diagnostic.message for diagnostic in result.diagnostics]
+    assert sorted(result.snapshot.entries) == ["provision.network.lab", "provision.node.web"]
+
+
+def test_a_backend_that_realizes_nothing_is_refused():
+    """The one fabrication the boundary already refuses."""
+
+    result = _apply_with(_RealizesNothing)
+
+    assert result.success is False
+    assert "runtime.backend-contract-invalid" in {diagnostic.code for diagnostic in result.diagnostics}
+
+
+def test_a_backend_cannot_invent_resources_the_scenario_never_authored():
+    result = _apply_with(_InventsResource)
+
+    assert result.success is False
+    assert "runtime.backend-contract-invalid" in {diagnostic.code for diagnostic in result.diagnostics}
+    assert _SMUGGLED_ADDRESS not in result.snapshot.entries
+
+
+def test_a_backend_cannot_misreport_what_kind_of_resource_it_realized():
+    result = _apply_with(_SubstitutesResourceType)
+
+    assert result.success is False
+    assert "runtime.backend-contract-invalid" in {diagnostic.code for diagnostic in result.diagnostics}
+    realized = {entry.resource_type for entry in result.snapshot.entries.values()}
+    assert "node" not in realized or realized == {"network", "node"}
+
+
+def test_a_backend_cannot_mutate_state_while_reporting_no_changes():
+    result = _apply_with(_UnderReportsChanges)
+
+    assert result.success is False
+    assert "runtime.backend-contract-invalid" in {diagnostic.code for diagnostic in result.diagnostics}
+    assert result.snapshot.entries == {}
+
+
+def test_a_backend_cannot_file_provisioning_results_under_another_domain():
+    result = _apply_with(_ForgesDomain)
+
+    assert result.success is False
+    assert "runtime.backend-contract-invalid" in {diagnostic.code for diagnostic in result.diagnostics}
+    domains = {getattr(entry.domain, "value", entry.domain) for entry in result.snapshot.entries.values()}
+    assert "evaluation" not in domains