ADC-681: resolve Reflux through the public AMR provider protocol

CHANGELOG.md

@@ -18,6 +18,9 @@ Format: [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); versioning
 
 ### Changed
 
+- External AMR `Reflux` components now use the normalized public provider route from
+  `AMR(..., reflux=...)` through resolve, compiled provenance and transactional native
+  installation; the builtin flux-register kernel follows the same reported contract.
 - AMR checkpoint capability reports now distinguish same-rank bit-identical replay from
   non-bit-identical rank-count rematerialization with Dense persisted histories. The M3 gate
   executes the persisted two-rank to one-rank restart proof. The explicit `RegridOnRestart()`

docs/design/SPECIFICATION_TECHNIQUE_FINALE_POPS_ARCHITECTURE.md

@@ -606,22 +606,24 @@ Les builtins de `pops.lib.amr` et les composants externes implémentent le même
 provider. Un composant externe est sélectionné sans callback Python :
 
 ```python
-from pops.amr import ClusteringProvider, TaggerProvider
+from pops.amr import ClusteringProvider, RefluxProvider, TaggerProvider
 
 layout = AMR(
     ...,
     tagger=TaggerProvider(component=my_tagger),
     clustering=ClusteringProvider(component=my_clustering),
+    reflux=RefluxProvider(component=my_reflux),
 )
 resolved = pops.resolve(
     pops.validate(case),
     layout=layout,
-    components=(my_tagger, my_clustering),
+    components=(my_tagger, my_clustering, my_reflux),
 )
 ```
 
-Les deux valeurs doivent référencer un exact `pops.external.ExternalComponent` portant
-respectivement l'interface générée `Tagger` ou `Clustering`. Le même objet exact doit être fourni à
+Les trois valeurs doivent référencer un exact `pops.external.ExternalComponent` portant
+respectivement l'interface générée `Tagger`, `Clustering` ou `Reflux`. Le même objet exact doit
+être fourni à
 `resolve(components=...)`; son identité de manifest, son interface et sa version traversent
 `resolve -> compile -> bind`. Le manifest doit déclarer une classification déterministe `bitwise` ou
 `reproducible`, car chaque rang doit produire la même hiérarchie. Un `Tagger` déclare en plus une
@@ -1426,11 +1428,12 @@ d'échec entre rangs, puis applique seul périodicité, masque de couverture, r
 publication transactionnelle. La présence et le contrat exact du provider sont également comparés
 entre rangs avant toute exécution.
 
-Cette tranche ne publie pas encore la sélection `Reflux` dans la résolution normalisée des providers
-AMR : le seam d'installation demeure interne et les configurations publiques continuent donc
-d'utiliser le kernel builtin. La qualification initiale de l'adaptateur reste limitée à la cible 2D,
-`float64`, CPU avec stockage hôte. Le chemin n'est pas encore prouvé par compilation native, exécution
-MPI avec un composant externe, mesure de conservation ni backend GPU.
+La sélection `AMR(..., reflux=RefluxProvider(component))` traverse désormais la même résolution
+normalisée, identité de provider, artifact et transaction d'installation que `Tagger` et
+`Clustering`. Sans sélection explicite, `FluxRegisterReflux` décrit le kernel builtin par le même
+protocole et apparaît dans le même rapport de providers. La qualification initiale de l'adaptateur
+reste limitée à la cible 2D, `float64`, CPU avec stockage hôte. Le chemin n'est pas encore prouvé par
+exécution MPI avec un composant externe, mesure de conservation ni backend GPU.
 
 Les champs sémantiques inconnus, capacités sans preuve, collisions d'identité et entry points manquants
 sont refusés. Un vieux manifest n'est pas « réparé » silencieusement.

python/pops/amr/__init__.py

@@ -36,6 +36,7 @@
     AMRProviderLoweringContext,
     amr_provider_binding_identity,
     ClusteringProvider,
+    RefluxProvider,
     ResolvedAMRProviderBinding,
     TaggerProvider,
 )
@@ -66,6 +67,7 @@
     "PatchLayout",
     "PreparedHierarchyNativeLowering",
     "PreparedHierarchyNativeProvider",
+    "RefluxProvider",
     "ResolvedAMRProviderBinding",
     "Tag",
     "TaggerProvider",

python/pops/amr/_resolution.py

@@ -625,6 +625,7 @@ def resolve_amr_authorities(
     load_balance: Any,
     tagger: Any,
     clustering: Any,
+    reflux: Any,
     context: AMRResolutionContext,
 ) -> ResolvedAMRAuthorities:
     """Resolve every adaptive-layout concern exactly once from its owning declaration."""
@@ -643,7 +644,7 @@ def resolve_amr_authorities(
                 raise TypeError("AMR %s authority must implement %s()" % (slot, method))
     if type(context) is not AMRResolutionContext:
         raise TypeError("AMR resolution requires an AMRResolutionContext")
-    providers = (tagger, clustering)
+    providers = (tagger, clustering, reflux)
     for value in providers:
         for method in ("inspect", "resolve_references", "lower_amr_provider"):
             if not callable(getattr(value, method, None)):
@@ -684,10 +685,11 @@ def resolve_amr_authorities(
         if lowered.role in provider_bindings:
             raise ValueError("AMR provider roles must be unique")
         provider_bindings[lowered.role] = lowered.data
-    if set(provider_bindings) != {"clustering", "tagger"}:
-        raise ValueError("AMR resolution requires exact clustering and tagger provider roles")
+    if set(provider_bindings) != {"clustering", "tagger", "reflux"}:
+        raise ValueError(
+            "AMR resolution requires exact clustering, tagger and reflux provider roles")
     provider_bindings = {
-        role: provider_bindings[role] for role in ("clustering", "tagger")
+        role: provider_bindings[role] for role in ("clustering", "tagger", "reflux")
     }
     resolved_hierarchy = _hierarchy(
         hierarchy,

python/pops/amr/providers.py

@@ -423,6 +423,64 @@ def runtime_binding_data(self) -> dict[str, Any]:
     canonical_identity = runtime_binding_data
 
 
+@dataclass(frozen=True, slots=True)
+class RefluxProvider:
+    """Bind one external local Reflux table to the conservative AMR transition."""
+
+    component: Any
+    __pops_ir_immutable__ = True
+
+    def __post_init__(self) -> None:
+        from pops import interfaces
+
+        _external_component(
+            self.component,
+            interface=interfaces.Reflux,
+            where="RefluxProvider.component",
+        )
+
+    def resolve_references(self, resolver: Any) -> RefluxProvider:
+        if not callable(resolver):
+            raise TypeError("RefluxProvider.resolve_references requires a callable resolver")
+        return self
+
+    def require_component_inputs(self, components: Any) -> None:
+        _require_component(self.component, components, where="RefluxProvider")
+
+    def lower_amr_provider(
+        self, context: AMRProviderLoweringContext,
+    ) -> ResolvedAMRProviderBinding:
+        """Authenticate the component, hierarchy layout and Program clock."""
+        if type(context) is not AMRProviderLoweringContext:
+            raise TypeError("RefluxProvider requires an AMRProviderLoweringContext")
+        self.require_component_inputs(context.components)
+        data = {
+            **self.runtime_binding_data(),
+            "layout_identity": context.layout_identity,
+            "clock_identity": context.clock_identity,
+        }
+        data["provider_identity"] = amr_provider_binding_identity("reflux", data)
+        return ResolvedAMRProviderBinding("reflux", data)
+
+    def runtime_binding_data(self) -> dict[str, Any]:
+        from pops import interfaces
+
+        data = {
+            "schema_version": 1,
+            "provider_type": "external_amr_reflux",
+            "runtime_installation": {
+                "schema_version": 1,
+                "protocol": "external_component",
+            },
+            **_component_binding(self.component, interfaces.Reflux),
+        }
+        data["provider_identity"] = make_identity("amr-reflux-provider", data).token
+        return data
+
+    inspect = runtime_binding_data
+    canonical_identity = runtime_binding_data
+
+
 @dataclass(frozen=True, slots=True)
 class _AMRRuntimeInterfaceProtocol:
     """Native-interface-owned validation and installation route."""
@@ -598,6 +656,26 @@ def validate_installed_capability(
                 "external AMR Tagger lacks its exact graph/capability/clock contract")
 
 
+@dataclass(frozen=True, slots=True)
+class _RefluxRuntimeInterfaceProtocol(_AMRRuntimeInterfaceProtocol):
+    """The local Reflux callback is qualified by the accepted Program clock."""
+
+    def validate_resolved_capability(
+        self, binding: Mapping[str, Any], resolved_tagging_identity: str | None,
+    ) -> None:
+        del resolved_tagging_identity
+        if not isinstance(binding.get("clock_identity"), str) \
+                or not binding["clock_identity"]:
+            raise ValueError("AMR Reflux lacks its exact Program clock authority")
+
+    def validate_installed_capability(
+        self, binding: Mapping[str, Any], installed: Any,
+        resolved_tagging_identity: str | None,
+    ) -> None:
+        del installed
+        self.validate_resolved_capability(binding, resolved_tagging_identity)
+
+
 def _runtime_interface_key(value: Any) -> tuple[Any, ...]:
     if not isinstance(value, Mapping):
         raise TypeError("AMR provider binding has no native-interface protocol")
@@ -632,6 +710,12 @@ def _runtime_interface_protocols() -> dict[tuple[Any, ...], _AMRRuntimeInterface
             builtin_provider_id="pops.lib.amr::symbolic_tagger",
             component_installer="_install_amr_tagger_component",
         ),
+        _RefluxRuntimeInterfaceProtocol(
+            role="reflux",
+            native_interface=interfaces.Reflux.to_data(),
+            builtin_provider_id="pops.lib.amr::flux_register_reflux",
+            component_installer="_install_amr_reflux_component",
+        ),
     )
     return {_runtime_interface_key(row.native_interface): row for row in protocols}
 
@@ -933,6 +1017,7 @@ def prepare_amr_provider_installation(
     "amr_provider_binding_identity",
     "ClusteringProvider",
     "PreparedAMRProviderNativeConfig",
+    "RefluxProvider",
     "ResolvedAMRProviderBinding",
     "TaggerProvider",
     "validate_amr_provider_binding",

python/pops/codegen/_amr_plan_validation.py

@@ -81,8 +81,9 @@ def validate_amr_authorities(plan: Any) -> None:
             or plan.bootstrap_plan.initial_identity != plan.initial_condition_plan.identity:
         raise ValueError("ResolvedSimulationPlan bootstrap does not authenticate AMR authorities")
     providers = plan.amr_providers
-    if tuple(providers) != ("clustering", "tagger"):
-        raise ValueError("AMR plan requires exact clustering and tagger provider bindings")
+    if tuple(providers) != ("clustering", "tagger", "reflux"):
+        raise ValueError(
+            "AMR plan requires exact clustering, tagger and reflux provider bindings")
     # Component inputs deliberately admit both source authorities and already-compiled
     # artifacts.  Their representations differ, but both expose the same authenticated
     # projection protocol.  Index that projection instead of reaching through the source-only

python/pops/layouts/__init__.py

@@ -396,6 +396,7 @@ def __init__(
         load_balance: Any = None,
         tagger: Any = None,
         clustering: Any = None,
+        reflux: Any = None,
     ) -> None:
         # Structural snapshots consume ``options()``.  Keeping authorities private prevents the
         # generic snapshotter from recursively treating Schedule implementation helpers as public
@@ -407,15 +408,22 @@ def __init__(
         self._transfer = transfer
         self._execution = execution
         self._patch_layout = PatchLayout() if patch_layout is None else patch_layout
-        if load_balance is None or tagger is None or clustering is None:
-            from pops.lib.amr import BergerRigoutsos, SpaceFillingCurve, SymbolicTagger
+        if load_balance is None or tagger is None or clustering is None or reflux is None:
+            from pops.lib.amr import (
+                BergerRigoutsos,
+                FluxRegisterReflux,
+                SpaceFillingCurve,
+                SymbolicTagger,
+            )
 
             load_balance = SpaceFillingCurve() if load_balance is None else load_balance
             tagger = SymbolicTagger() if tagger is None else tagger
             clustering = BergerRigoutsos() if clustering is None else clustering
+            reflux = FluxRegisterReflux() if reflux is None else reflux
         self._load_balance = load_balance
         self._tagger = tagger
         self._clustering = clustering
+        self._reflux = reflux
 
     @property
     def grid(self) -> Any:
@@ -457,6 +465,10 @@ def tagger(self) -> Any:
     def clustering(self) -> Any:
         return self._clustering
 
+    @property
+    def reflux(self) -> Any:
+        return self._reflux
+
     def _validate_authorities(self) -> None:
         authorities = {
             "hierarchy": self.hierarchy, "tagging": self.tagging,
@@ -468,6 +480,7 @@ def _validate_authorities(self) -> None:
         _load_balance_data(self.load_balance)
         _provider_data(self.tagger, "tagger")
         _provider_data(self.clustering, "clustering")
+        _provider_data(self.reflux, "reflux")
         for method in ("validate", "capabilities", "requirements", "options", "to_dict"):
             if not callable(getattr(self.grid, method, None)):
                 raise TypeError("AMR.grid must implement %s()" % method)
@@ -516,6 +529,7 @@ def options(self) -> dict[str, Any]:
             "load_balance": _load_balance_data(self.load_balance),
             "tagger": self.tagger.inspect(),
             "clustering": self.clustering.inspect(),
+            "reflux": self.reflux.inspect(),
         }
 
     def _summary(self) -> str:
@@ -567,6 +581,7 @@ def resolved(value: Any) -> Any:
             load_balance=self.load_balance,
             tagger=self.tagger.resolve_references(resolved),
             clustering=self.clustering.resolve_references(resolved),
+            reflux=self.reflux.resolve_references(resolved),
         )
 
     def resolve_amr_authorities(self, context: Any) -> Any:
@@ -583,6 +598,7 @@ def resolve_amr_authorities(self, context: Any) -> Any:
             load_balance=self.load_balance,
             tagger=self.tagger,
             clustering=self.clustering,
+            reflux=self.reflux,
             context=context,
         )
 

python/pops/lib/amr/__init__.py

@@ -373,6 +373,58 @@ def runtime_binding_data(self) -> dict[str, Any]:
     canonical_identity = runtime_binding_data
 
 
+@dataclass(frozen=True, slots=True)
+class FluxRegisterReflux:
+    """Builtin conservative flux-register correction through the Reflux provider protocol."""
+
+    __pops_ir_immutable__: ClassVar[bool] = True
+
+    def resolve_references(self, resolver: Any) -> FluxRegisterReflux:
+        if not callable(resolver):
+            raise TypeError("FluxRegisterReflux.resolve_references requires a callable resolver")
+        return self
+
+    def require_component_inputs(self, components: Any) -> None:
+        del components
+
+    def lower_amr_provider(self, context: Any) -> Any:
+        from pops.amr.providers import (
+            AMRProviderLoweringContext,
+            ResolvedAMRProviderBinding,
+            amr_provider_binding_identity,
+        )
+
+        if type(context) is not AMRProviderLoweringContext:
+            raise TypeError("FluxRegisterReflux requires an AMRProviderLoweringContext")
+        self.require_component_inputs(context.components)
+        data = {
+            **self.runtime_binding_data(),
+            "layout_identity": context.layout_identity,
+            "clock_identity": context.clock_identity,
+        }
+        data["provider_identity"] = amr_provider_binding_identity("reflux", data)
+        return ResolvedAMRProviderBinding("reflux", data)
+
+    def runtime_binding_data(self) -> dict[str, Any]:
+        from pops import interfaces
+
+        data = {
+            "schema_version": 1,
+            "provider_type": "builtin_amr_reflux",
+            "runtime_installation": {
+                "schema_version": 1,
+                "protocol": "builtin",
+            },
+            "provider_id": "pops.lib.amr::flux_register_reflux",
+            "native_interface": interfaces.Reflux.to_data(),
+        }
+        data["provider_identity"] = make_identity("amr-reflux-provider", data).token
+        return data
+
+    inspect = runtime_binding_data
+    canonical_identity = runtime_binding_data
+
+
 @dataclass(frozen=True, slots=True)
 class BergerRigoutsos:
     """Builtin clustering provider with intrinsic validated algorithm controls."""
@@ -459,6 +511,7 @@ def runtime_binding_data(self) -> dict[str, Any]:
     "DivergencePreservingFace",
     "EllipticRecompute",
     "FaceTransfer",
+    "FluxRegisterReflux",
     "LinearTimeInterpolation",
     "Knapsack",
     "NodeTransfer",

python/pops/runtime/_runtime_authorities.py

@@ -511,8 +511,10 @@ def _install_amr_provider_authorities(engine: Any, install_plan: Any) -> None:
     """Install every AMR provider through its authority-carried runtime protocol."""
 
     providers = install_plan.amr_providers
-    if not isinstance(providers, Mapping) or tuple(providers) != ("clustering", "tagger"):
-        raise ValueError("adaptive runtime requires exact clustering and tagger providers")
+    if not isinstance(providers, Mapping) \
+            or tuple(providers) != ("clustering", "tagger", "reflux"):
+        raise ValueError(
+            "adaptive runtime requires exact clustering, tagger and reflux providers")
     native = getattr(engine, "_s", None)
     from pops.amr.providers import prepare_amr_provider_installation
     from pops.runtime._component_execution_context import component_execution_data