fix(codegen): bind qualified flux provider packs (ADC-682)

include/pops/numerics/fv/flux_interfaces.hpp

@@ -96,6 +96,7 @@ struct QualifiedProviderRequirement {
   const char* layout;
   const char* value_kind;
   const char* producer;
+  bool available;
   int storage_slot;
 };
 

python/pops/codegen/_compiler_lowering.py

@@ -12,6 +12,7 @@ class _CompilerEmitter(Protocol):
     """Minimal executable half of a compiler lowering."""
 
     def check(self) -> object: ...
+    def __pops_bind_component_provider_packs__(self, packs: Any) -> None: ...
     def __pops_native_loader_source__(
         self, *, name: Any = None, target: str = "system",
         hoist_reciprocals: bool = False,
@@ -26,6 +27,14 @@ class CompilerLowering:
     source_module: Module
     facade: object
 
+    def bind_component_provider_packs(self, packs: Any) -> None:
+        """Bind one resolved provider-pack authority before native source emission."""
+        result = self.emit_model.__pops_bind_component_provider_packs__(packs)
+        if result is not None:
+            raise TypeError(
+                "compiler provider-pack binding protocol must return None"
+            )
+
     def native_loader_source(
         self, *, name: Any = None, target: str = "system",
         hoist_reciprocals: bool = False,

python/pops/codegen/component_provider_packs.py

@@ -0,0 +1,104 @@
+"""Exact component-provider packs shared by every compiler entry route.
+
+The operator-first :class:`pops.model.Module` is the authority for provider identity.  Kernel
+emitters must not rediscover providers from the legacy auxiliary layout: this module resolves the
+full pack, every per-operator subset, and the physical-flux subset once and passes that immutable
+value through the explicit compiler-emitter protocol.
+"""
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Any
+
+from pops.model.provider_pack import (
+    ProviderPack,
+    build_operator_provider_pack,
+    build_provider_pack,
+)
+
+
+@dataclass(frozen=True, slots=True)
+class ComponentProviderPacks:
+    """One immutable provider resolution for a canonical Module."""
+
+    complete: ProviderPack
+    by_operator: Mapping[str, ProviderPack]
+    physical_flux: ProviderPack
+
+    def __post_init__(self) -> None:
+        if type(self.complete) is not ProviderPack:
+            raise TypeError("ComponentProviderPacks.complete must be an exact ProviderPack")
+        rows = dict(self.by_operator)
+        if any(not isinstance(name, str) or not name for name in rows):
+            raise TypeError(
+                "ComponentProviderPacks.by_operator keys must be non-empty strings"
+            )
+        if any(type(pack) is not ProviderPack for pack in rows.values()):
+            raise TypeError(
+                "ComponentProviderPacks.by_operator values must be exact ProviderPack values"
+            )
+        object.__setattr__(self, "by_operator", MappingProxyType(rows))
+        if type(self.physical_flux) is not ProviderPack:
+            raise TypeError(
+                "ComponentProviderPacks.physical_flux must be an exact ProviderPack"
+            )
+
+    def attach(self, target: Any) -> None:
+        """Attach compiler-owned immutable evidence to one emitter carrier.
+
+        Reattachment is idempotent and verifies byte-for-byte logical equality.  This is needed
+        because a facade and its private formula carrier are distinct Python objects but emit one
+        native package; neither may retain a different provider resolution.
+        """
+        values = {
+            "_component_provider_pack": self.complete,
+            "_component_provider_metadata": self.complete.to_data(),
+            "_component_operator_provider_packs": self.by_operator,
+            "_component_operator_provider_metadata": MappingProxyType({
+                name: pack.to_data() for name, pack in self.by_operator.items()
+            }),
+            "_component_flux_provider_pack": self.physical_flux,
+            "_component_flux_provider_metadata": self.physical_flux.to_data(),
+        }
+
+        def canonical(value: Any) -> Any:
+            if isinstance(value, ProviderPack):
+                return value.to_data()
+            if isinstance(value, Mapping):
+                return {
+                    key: canonical(item)
+                    for key, item in value.items()
+                }
+            return value
+
+        for name, value in values.items():
+            previous = getattr(target, name, None)
+            if previous is not None and canonical(previous) != canonical(value):
+                raise ValueError(
+                    "compiler emitter retained a conflicting component-provider pack"
+                )
+            object.__setattr__(target, name, value)
+
+
+def resolve_component_provider_packs(module: Any) -> ComponentProviderPacks:
+    """Resolve all exact provider packs from one canonical Module authority."""
+    complete = build_provider_pack(module)
+    by_operator = {
+        operator.name: build_operator_provider_pack(module, operator)
+        for operator in module.operator_registry()
+    }
+    flux_requirements = []
+    for operator in module.operator_registry():
+        if operator.kind == "grid_operator":
+            flux_requirements.extend(by_operator[operator.name])
+    physical_flux = complete.select(flux_requirements)
+    return ComponentProviderPacks(
+        complete=complete,
+        by_operator=by_operator,
+        physical_flux=physical_flux,
+    )
+
+
+__all__ = ["ComponentProviderPacks", "resolve_component_provider_packs"]

python/pops/codegen/module_emit_brick.py

@@ -241,8 +241,10 @@ def roles_init(roles: Any) -> Any:
             contract["representation"], contract["centering"], contract["unit"] or "",
             contract["layout"], contract["value_kind"] or "", provider["producer"] or "",
         ]
-        S.append("    {%s, %d}," %
-                 (", ".join(json.dumps(value) for value in values), provider["slot"]))
+        availability = "true" if provider["availability"] else "false"
+        S.append("    {%s, %s, %d}," %
+                 (", ".join(json.dumps(value) for value in values),
+                  availability, provider["slot"]))
     S.append("  }};")
     if rt_member:  # member pops::RuntimeParams params{count, {defaults}} (P7-b)
         S.append(rt_member.rstrip("\n"))

python/pops/codegen/module_lowering.py

@@ -20,7 +20,6 @@
 
 from __future__ import annotations
 
-from types import MappingProxyType
 from collections.abc import Iterable, Mapping
 from typing import Any, cast
 
@@ -75,30 +74,13 @@ def _body_for_state(body: Any) -> Any:
     # Preserve the canonical source-Module identity across the internal facade lowering. The
     # resulting CompiledModel authenticates this scalar hash; it never retains ``module`` itself.
     object.__setattr__(m, "_compile_source_module_hash", module.module_hash())
-    from pops.model.provider_pack import (  # noqa: PLC0415
-        build_operator_provider_pack,
-        build_provider_pack,
+    from pops.codegen.component_provider_packs import (  # noqa: PLC0415
+        resolve_component_provider_packs,
     )
 
-    provider_pack = build_provider_pack(module)
-    object.__setattr__(m, "_component_provider_pack", provider_pack)
-    object.__setattr__(m, "_component_provider_metadata", provider_pack.to_data())
-    operator_provider_packs = {
-        operator.name: build_operator_provider_pack(module, operator)
-        for operator in module.operator_registry()
-    }
-    object.__setattr__(m, "_component_operator_provider_packs",
-                       MappingProxyType(operator_provider_packs))
-    object.__setattr__(m, "_component_operator_provider_metadata", MappingProxyType({
-        name: pack.to_data() for name, pack in operator_provider_packs.items()
-    }))
-    flux_keys = []
-    for operator in module.operator_registry():
-        if operator.kind == "grid_operator":
-            flux_keys.extend(operator_provider_packs[operator.name])
-    flux_provider_pack = provider_pack.select(flux_keys)
-    object.__setattr__(m, "_component_flux_provider_pack", flux_provider_pack)
-    object.__setattr__(m, "_component_flux_provider_metadata", flux_provider_pack.to_data())
+    m.__pops_bind_component_provider_packs__(
+        resolve_component_provider_packs(module)
+    )
     # The facade is a lowering view of THIS Module, not a newly declared model. Re-anchor its empty
     # backing model before the first declaration so every derived operator registry retains the
     # Module's exact authoring authority. Without this, owner-qualified Program nodes would be
@@ -191,7 +173,7 @@ def _declare_aux(nm: Any, key: Any) -> None:
         coverage_rows.append(LoweringCoverageRow(
             "module:%s:eigenvalues" % module.name, "documentary"))
 
-    for key in provider_pack:
+    for key in m._component_provider_pack:
         key_data = key.to_data()
         stable_key = "%s/%s/%s" % (
             key_data["space_kind"], key_data["space_name"], key_data["component"])
@@ -441,6 +423,11 @@ def lower_and_validate(model: Any, facade: Any = None, state_space: Any = None)
         lowering = require_compiler_lowering(model)
         if diagnostic_facade is None:
             diagnostic_facade = lowering.facade
+        from pops.codegen.component_provider_packs import resolve_component_provider_packs
+
+        lowering.bind_component_provider_packs(
+            resolve_component_provider_packs(lowering.source_module)
+        )
         states = lowering.source_module.state_spaces()
         if len(states) > 1:
             emit_model = _module_to_model(

python/pops/model/provider_pack.py

@@ -256,6 +256,50 @@ def select_spaces(self, *, owner_qid: str,
                 (owner_qid, sorted(missing)))
         return self.select(keys)
 
+    def select_components(
+        self,
+        *,
+        owner_qid: str,
+        spaces: Iterable[tuple[str, str]],
+        components: Iterable[str],
+    ) -> ProviderPack:
+        """Select exact components from declared spaces without a bare-name fallback.
+
+        Component spelling is only a filter inside the already-qualified owner/space set.  A
+        missing component or the same spelling in two selected spaces is rejected rather than
+        guessed, so an operator that needs one of two homonymous fields must qualify its input
+        space more narrowly.
+        """
+        _non_empty(owner_qid, "ProviderPack selection owner_qid")
+        requested_spaces = set(spaces)
+        requested_components = tuple(components)
+        if any(not isinstance(name, str) or not name for name in requested_components):
+            raise TypeError(
+                "ProviderPack components must contain non-empty strings"
+            )
+        if len(set(requested_components)) != len(requested_components):
+            raise ValueError("ProviderPack components contains a duplicate")
+        candidates = [
+            key for key in self
+            if key.owner_qid == owner_qid
+            and (key.space_kind, key.space_name) in requested_spaces
+        ]
+        selected = []
+        for component in requested_components:
+            matches = [key for key in candidates if key.component == component]
+            if not matches:
+                raise MissingInputProvider(
+                    "missing component %r in qualified provider spaces %r for owner %r"
+                    % (component, sorted(requested_spaces), owner_qid)
+                )
+            if len(matches) != 1:
+                raise MissingInputProvider(
+                    "ambiguous component %r in qualified provider spaces %r for owner %r"
+                    % (component, sorted(requested_spaces), owner_qid)
+                )
+            selected.append(matches[0])
+        return self.select(selected)
+
     def to_data(self) -> dict[str, Any]:
         rows = []
         for key in sorted(self._entries):
@@ -363,7 +407,16 @@ def build_operator_provider_pack(module: Any, operator: Any) -> ProviderPack:
             spaces.append(("field", input_space.name))
     if not spaces:
         return ProviderPack(capacity=full.capacity)
-    return full.select_spaces(owner_qid=str(module.owner_path.canonical()), spaces=spaces)
+    owner_qid = str(module.owner_path.canonical())
+    requirements = getattr(operator, "requirements", {})
+    required_components = requirements.get("aux", ())
+    if required_components:
+        return full.select_components(
+            owner_qid=owner_qid,
+            spaces=spaces,
+            components=required_components,
+        )
+    return full.select_spaces(owner_qid=owner_qid, spaces=spaces)
 
 
 __all__ = ["ComponentKey", "ComponentContract", "ProviderEntry", "ProviderPack",

python/pops/physics/_authoring_view.py

@@ -30,7 +30,23 @@ def _aux_name_set(self) -> Any:
     def _aux_requirements(self, exprs: Any) -> Any:
         """{'aux': [...]} of the aux fields the expressions read, or {} if none."""
         aux_set = self._aux_name_set()
-        read = sorted(_dependencies(exprs) & aux_set)
+        dependencies = _dependencies(exprs)
+        pending = [name for name in dependencies if name in self.prim_defs]
+        expanded = set(dependencies)
+        visited = set()
+        while pending:
+            name = pending.pop()
+            if name in visited:
+                continue
+            visited.add(name)
+            nested = _dependencies((self.prim_defs[name],))
+            expanded.update(nested)
+            pending.extend(
+                dependency
+                for dependency in nested
+                if dependency in self.prim_defs and dependency not in visited
+            )
+        read = sorted(expanded & aux_set)
         return {"aux": read} if read else {}
 
     def state_space(self, name: str = "U") -> Any:
@@ -85,25 +101,68 @@ def operator_registry(self, state_name: str = "U") -> Any:
         reg = _model.OperatorRegistry(owner=self.owner_path)
         state = self.state_space(state_name)
         fields = self.field_space()
-        aux_set = self._aux_name_set()
 
         def reads_fields(exprs: Any) -> bool:
-            return bool(_dependencies(exprs) & aux_set)
+            return bool(self._aux_requirements(exprs))
+
+        stability_exprs = [
+            *self._eig.get("x", ()),
+            *self._eig.get("y", ()),
+        ]
+        if self._wave_speeds is not None:
+            stability_exprs.extend(self._wave_speeds["x"])
+            stability_exprs.extend(self._wave_speeds["y"])
+        if self._ws_jacobian is not None and self._ws_jacobian["rows"] is not None:
+            for direction in ("x", "y"):
+                stability_exprs.extend(
+                    expression
+                    for row in self._ws_jacobian["rows"][direction]
+                    for expression in row
+                )
+        if self._roe_rows is not None:
+            stability_exprs.extend(self._roe_rows["x"])
+            stability_exprs.extend(self._roe_rows["y"])
+        if self._roe_jacobian is not None:
+            for direction in ("x", "y"):
+                stability_exprs.extend(
+                    expression
+                    for row in self._roe_jacobian[direction]
+                    for expression in row
+                )
 
         # Flux divergence (grid_operator: State -> Rate(State)).
         if self._flux:
+            exprs = [
+                *self._flux.get("x", ()),
+                *self._flux.get("y", ()),
+                *stability_exprs,
+            ]
+            rf = reads_fields(exprs)
             reg.register(_model.Operator(
                 "flux_default", "grid_operator",
-                _model.Signature([state], _model.Rate(state)),
+                _model.Signature([state, fields] if rf else [state],
+                                 _model.Rate(state)),
                 capabilities={"local": False, "linear": False, "produces_rate": True,
                               "requires_ghosts": 1, "supports_device": True,
-                              "default": True},
+                              "requires_fields": rf, "default": True},
+                requirements=self._aux_requirements(exprs),
                 source=None))
         for nm in sorted(self._flux_terms):
+            term = self._flux_terms[nm]
+            exprs = [
+                *term.get("x", ()),
+                *term.get("y", ()),
+                *stability_exprs,
+            ]
+            rf = reads_fields(exprs)
             reg.register(_model.Operator(
-                nm, "grid_operator", _model.Signature([state], _model.Rate(state)),
+                nm, "grid_operator",
+                _model.Signature([state, fields] if rf else [state],
+                                 _model.Rate(state)),
                 capabilities={"local": False, "linear": False, "produces_rate": True,
-                              "requires_ghosts": 1, "supports_device": True},
+                              "requires_ghosts": 1, "supports_device": True,
+                              "requires_fields": rf},
+                requirements=self._aux_requirements(exprs),
                 source=None))
 
         # Local sources (local_source: State[, Fields] -> Rate(State)).