ADC-684: fail closed on unowned runtime actions

CHANGELOG.md

@@ -18,6 +18,10 @@ Format: [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); versioning
 
 ### Changed
 
+- Native `RuntimeInstance` installation now maps each single-layout halo requirement to the exact
+  compiled block ghost allocation and fails before backend inspection when a planned buffer,
+  cross-memory fence, or clock join has no execution owner. ConsumerGraph collectives retain their
+  separate transactional owner.
 - `Program.cadence(substeps=..., stride=...)` now authors the native global cadence as immutable,
   identity-bearing Program data and installs it before the Uniform or AMR runtime freezes.
 - `AsyncScientificOutput` now accepts fields, diagnostics, or both on one exact schedule. Diagnostic

docs/design/runtime_instance_planning_contract.md

@@ -85,10 +85,16 @@ complete bundle is retained in the array-free `RuntimeInstance.inspect()` report
 assumptions remain reviewable rather than becoming hidden installation state.
 Single-layout providers additionally require the exact ordered block/layout call projection,
 layout-qualified halos, and the absence of unconsumed Transfer or mapping-provider routes before
-constructing their sole native engine.
+constructing their sole native engine. Every planned halo is mapped back to its owning compiled
+block and its derived depth must fit the authenticated spatial ghost allocation; a second runtime
+ghost-depth knob does not exist.
 The multi-layout Uniform provider likewise authenticates ordered block/layout calls and the exact
 mapping-provider set backing its materialized Transfers before constructing child engines. It
 refuses non-empty runtime halo plans until an explicit per-layout halo scheduler exists.
+The provider boundary also refuses planned buffer allocations, cross-memory fences, and clock
+joins while no native owner exists for those action classes. They are never accepted as advisory
+metadata. Consumer-owned collectives remain outside this refusal because `ConsumerGraph` lowers
+and executes them transactionally through its own authenticated plan.
 
 For an accepted step, successful native finalization is an irreversible `native_finalized`
 boundary. The instance commits the engine state, accepted cursor set and consumer receipts across

python/pops/runtime/_runtime_executor.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
+from collections.abc import Mapping
 from typing import Any, cast
 
 from pops.codegen._plans import require_install_plan
@@ -188,6 +189,35 @@ def _require_runtime_determinism(
     runtime_plan.determinism.require_assumptions(actual)
 
 
+def _require_supported_runtime_actions(runtime_plan: Any) -> None:
+    """Refuse derived actions for which no native execution owner exists yet."""
+    unsupported = (
+        ("buffer allocations", runtime_plan.resources.buffers),
+        ("cross-memory fences", runtime_plan.communication.fences),
+        ("clock joins", runtime_plan.communication.clock_joins),
+    )
+    for label, rows in unsupported:
+        if rows:
+            raise NotImplementedError(
+                "native RuntimeInstance has no execution owner for planned %s" % label
+            )
+
+
+def _compiled_spatial_ghost_depth(block: Any) -> int:
+    spatial = getattr(block, "spatial", None)
+    value = (
+        spatial.get("ghost_depth")
+        if isinstance(spatial, Mapping)
+        else getattr(spatial, "ghost_depth", None)
+    )
+    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
+        raise TypeError(
+            "compiled block %r has no exact positive spatial ghost depth"
+            % getattr(block, "name", None)
+        )
+    return value
+
+
 def _require_single_layout_runtime_plan(plan: Any, runtime_plan: Any) -> None:
     """Require the exact call/layout projection consumed by one native engine."""
     layout_plan = plan.artifact.layout_plan
@@ -211,6 +241,24 @@ def _require_single_layout_runtime_plan(plan: Any, runtime_plan: Any) -> None:
         raise ValueError("single-layout native provider cannot consume mapping providers")
     if any(row.layout_id != layout_id for row in runtime_plan.communication.halos):
         raise ValueError("RuntimePlanBundle halo differs from the installed single layout")
+    calls = {row.identity.token: row for row in runtime_plan.calls}
+    if len(calls) != len(runtime_plan.calls):
+        raise ValueError("RuntimePlanBundle contains duplicate RuntimeCall identities")
+    block_names = {subject_id: name for name, (subject_id, _) in assignments.items()}
+    compiled = {row.name: row for row in plan.artifact.blocks}
+    if set(compiled) != set(assignments):
+        raise ValueError("compiled block set differs from the single-layout plan")
+    for halo in runtime_plan.communication.halos:
+        call = calls.get(halo.call_id)
+        if call is None or call.block_id not in block_names:
+            raise ValueError("RuntimePlanBundle halo has no installed block owner")
+        block = compiled[block_names[call.block_id]]
+        available = _compiled_spatial_ghost_depth(block)
+        if halo.depth > available:
+            raise ValueError(
+                "RuntimePlanBundle halo depth %d exceeds compiled block %r ghost depth %d"
+                % (halo.depth, block.name, available)
+            )
 
 
 class _UniformNativeProvider(RuntimeExecutorProvider):
@@ -343,6 +391,7 @@ def install_runtime_executor(install_plan: Any, runtime_plan: Any = None) -> Any
     from pops.runtime._runtime_planning import require_runtime_plan_bundle
 
     runtime_plan = require_runtime_plan_bundle(plan, runtime_plan)
+    _require_supported_runtime_actions(runtime_plan)
     native_facts = _native_runtime_facts()
     _require_runtime_determinism(plan, runtime_plan, native_facts)
     _require_supported_execution_context(plan, native_facts)

tests/python/unit/runtime/test_runtime_executor_context.py

@@ -156,7 +156,12 @@ def test_determinism_assumptions_are_rechecked_before_native_preflight(monkeypat
             {},
             make_identity("execution-context", {"test": "runtime-executor"}),
         ),
-        communication=SimpleNamespace(collectives=()),
+        resources=SimpleNamespace(buffers=()),
+        communication=SimpleNamespace(
+            collectives=(),
+            fences=(),
+            clock_joins=(),
+        ),
     )
     calls = []
 
@@ -204,9 +209,14 @@ def test_matching_runtime_determinism_assumptions_are_consumed():
 
 def _single_layout_projection():
     layout = SimpleNamespace(handle=SimpleNamespace(qualified_id="layout::primary"))
+    call_identity = SimpleNamespace(token="runtime-call::fluid")
+    compiled_block = SimpleNamespace(
+        name="fluid",
+        spatial={"ghost_depth": 2},
+    )
     plan = SimpleNamespace(
         artifact=SimpleNamespace(
-            blocks=(SimpleNamespace(name="fluid"),),
+            blocks=(compiled_block,),
             layout_plan=SimpleNamespace(
                 layouts=(layout,),
                 assignments=(
@@ -223,12 +233,25 @@ def _single_layout_projection():
         )
     )
     runtime_plan = SimpleNamespace(
-        calls=(SimpleNamespace(block_id="block::fluid", layout_id="layout::primary"),),
+        calls=(
+            SimpleNamespace(
+                identity=call_identity,
+                block_id="block::fluid",
+                layout_id="layout::primary",
+            ),
+        ),
         communication=SimpleNamespace(
             transfers=(),
-            halos=(SimpleNamespace(layout_id="layout::primary"),),
+            halos=(
+                SimpleNamespace(
+                    call_id=call_identity.token,
+                    resource="state:u",
+                    layout_id="layout::primary",
+                    depth=2,
+                ),
+            ),
         ),
-        resources=SimpleNamespace(mapping_provider_ids=()),
+        resources=SimpleNamespace(mapping_provider_ids=(), buffers=()),
     )
     return plan, runtime_plan
 
@@ -248,6 +271,22 @@ def test_single_layout_provider_consumes_exact_call_and_halo_projection():
         executor._require_single_layout_runtime_plan(plan, runtime_plan)
 
 
+def test_single_layout_provider_refuses_halo_deeper_than_compiled_storage():
+    plan, runtime_plan = _single_layout_projection()
+    runtime_plan.communication.halos[0].depth = 3
+
+    with pytest.raises(ValueError, match="exceeds compiled block.*ghost depth 2"):
+        executor._require_single_layout_runtime_plan(plan, runtime_plan)
+
+
+def test_single_layout_provider_requires_exact_compiled_halo_evidence():
+    plan, runtime_plan = _single_layout_projection()
+    plan.artifact.blocks[0].spatial = {}
+
+    with pytest.raises(TypeError, match="exact positive spatial ghost depth"):
+        executor._require_single_layout_runtime_plan(plan, runtime_plan)
+
+
 @pytest.mark.parametrize("transfers,providers,match", [
     ((object(),), (), "layout Transfers"),
     ((), ("pops://mapping/test",), "mapping providers"),
@@ -346,6 +385,68 @@ def test_multi_layout_provider_refuses_unconsumed_halo_plan():
         multi_executor._require_runtime_plan_projection(plan, runtime_plan, transfers)
 
 
+@pytest.mark.parametrize(
+    "resource_rows,fences,clock_joins,match",
+    [
+        ((object(),), (), (), "buffer allocations"),
+        ((), (object(),), (), "cross-memory fences"),
+        ((), (), (object(),), "clock joins"),
+    ],
+)
+def test_runtime_provider_refuses_planned_actions_without_native_owner(
+    resource_rows, fences, clock_joins, match
+):
+    runtime_plan = SimpleNamespace(
+        resources=SimpleNamespace(buffers=resource_rows),
+        communication=SimpleNamespace(
+            fences=fences,
+            clock_joins=clock_joins,
+            collectives=(object(),),
+        ),
+    )
+
+    with pytest.raises(NotImplementedError, match=match):
+        executor._require_supported_runtime_actions(runtime_plan)
+
+
+def test_consumer_collectives_are_not_claimed_by_runtime_action_gate():
+    runtime_plan = SimpleNamespace(
+        resources=SimpleNamespace(buffers=()),
+        communication=SimpleNamespace(
+            fences=(),
+            clock_joins=(),
+            collectives=(object(),),
+        ),
+    )
+
+    executor._require_supported_runtime_actions(runtime_plan)
+
+
+def test_unowned_runtime_action_fails_before_native_fact_probe(monkeypatch):
+    plan = SimpleNamespace()
+    runtime_plan = SimpleNamespace(
+        resources=SimpleNamespace(buffers=(object(),)),
+        communication=SimpleNamespace(
+            collectives=(),
+            fences=(),
+            clock_joins=(),
+        ),
+    )
+
+    def forbidden_native_facts():
+        raise AssertionError("native fact probe became reachable")
+
+    monkeypatch.setattr(executor, "require_install_plan", lambda value: value)
+    monkeypatch.setattr(
+        runtime_planning,
+        "require_runtime_plan_bundle",
+        lambda _plan, value: value,
+    )
+    monkeypatch.setattr(executor, "_native_runtime_facts", forbidden_native_facts)
+    with pytest.raises(NotImplementedError, match="buffer allocations"):
+        executor.install_runtime_executor(plan, runtime_plan)
+
+
 
 
 def test_before_step_transfer_cycle_captures_every_native_source_before_any_apply():