feat(lattice): wire durable auto config runtime

apps/predbat/gateway.py

@@ -1,3 +1,4 @@
+# cspell:ignore autoconfig
 """ESP32 Gateway MQTT component.
 
 Provides full inverter telemetry and control via the ESP32 gateway's
@@ -253,6 +254,7 @@ def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqt
         self._loop = None
         self._gateway_online = False
         self._last_telemetry_time = 0
+        self._lattice_gateway_stale = False
         self._last_plan_data = None
         self._last_plan_publish_time = 0
         # Entries and timezone of the last built plan, kept so the periodic re-publish
@@ -576,6 +578,8 @@ async def run(self, seconds, first):
                 self.log("Info: GatewayMQTT: Restarting MQTT listener task")
                 self._mqtt_task = asyncio.ensure_future(self._mqtt_loop())
 
+            await self._invalidate_stale_lattice_gateway()
+
             # Publish any queued plan from on_plan_executed hook
             if self._pending_plan:
                 plan_entries, tz = self._pending_plan
@@ -687,7 +691,14 @@ async def _handle_message(self, message):
 
         try:
             if topic == self.topic_status:
-                self._process_telemetry(message.payload)
+                lattice_runtime = self._active_lattice_runtime()
+                if lattice_runtime is None:
+                    self._process_telemetry(message.payload)
+                else:
+                    await self._process_lattice_telemetry(
+                        message.payload,
+                        lattice_runtime,
+                    )
             elif topic == self.topic_online:
                 payload = message.payload.decode("utf-8", errors="replace").strip()
                 was_online = self._gateway_online
@@ -700,11 +711,108 @@ async def _handle_message(self, message):
                         attributes=GATEWAY_ATTRIBUTE_TABLE.get("gateway_online", {}),
                         app="gateway",
                     )
+                    lattice_runtime = self._active_lattice_runtime()
+                    if lattice_runtime is not None:
+                        await self._set_lattice_gateway_liveness(
+                            lattice_runtime,
+                            self._gateway_online,
+                        )
         except Exception as e:
             self._error_count += 1
             self.log(f"Warn: GatewayMQTT: Error handling message on {topic}: {e}")
             self.log(f"Warn: {traceback.format_exc()}")
 
+    def _active_lattice_runtime(self):
+        """Return the enabled live Lattice runtime, if one is installed."""
+        runtime = getattr(self.base, "lattice_autoconfig_runtime", None)
+        if runtime is None or getattr(runtime, "enabled", False) is not True:
+            return None
+        if runtime.provider_active("predbat-gateway") is not True:
+            return None
+        return runtime
+
+    async def stop(self):
+        """Invalidate the long-lived provider before a component restart."""
+        runtime = self._active_lattice_runtime()
+        if runtime is not None:
+            await self._set_lattice_gateway_liveness(runtime, False)
+        await super().stop()
+
+    async def _set_lattice_gateway_liveness(self, runtime, online):
+        """Publish Gateway liveness without blocking the MQTT event loop."""
+        try:
+            await asyncio.to_thread(runtime.set_gateway_liveness, online)
+            return True
+        except Exception as e:
+            self._error_count += 1
+            self.log(f"Warn: GatewayMQTT: Lattice liveness update failed: {e}")
+            return False
+
+    async def _invalidate_stale_lattice_gateway(self):
+        """Invalidate one broker-connected provider after telemetry goes stale."""
+        runtime = self._active_lattice_runtime()
+        stale = self._last_telemetry_time and (time.time() - self._last_telemetry_time) >= _TELEMETRY_STALE_THRESHOLD
+        if runtime is None or not stale or self._lattice_gateway_stale:
+            return False
+        invalidated = await self._set_lattice_gateway_liveness(runtime, False)
+        if invalidated:
+            self._lattice_gateway_stale = True
+        return invalidated
+
+    def _synchronize_lattice_gateway_plan(self, plan):
+        """Install the pure plan's stable serial selection into Gateway state."""
+        selected_serials = tuple(getattr(plan, "selected_serials", ()))
+        discovered_serials = tuple(getattr(plan, "discovered_serials", ()))
+        if not selected_serials:
+            raise ValueError("Lattice Gateway plan selected no inverters")
+        suffix_to_serial = {_serial_suffix(serial): serial for serial in selected_serials}
+        if len(suffix_to_serial) != len(selected_serials):
+            raise ValueError("Lattice Gateway plan contains colliding serial suffixes")
+        self._suffix_to_serial = suffix_to_serial
+        self._configured_inverter_serials = frozenset(discovered_serials)
+        self._configured_ev_chargers = frozenset()
+        self._auto_configured = True
+
+    async def _process_lattice_telemetry(self, data, runtime):
+        """Decode, publish, and activate one Lattice-managed Gateway status."""
+        try:
+            status = pb.GatewayStatus()
+            status.ParseFromString(data)
+        except Exception as e:
+            self._error_count += 1
+            self.log(f"Warn: GatewayMQTT: Failed to decode telemetry: {e}")
+            return
+
+        self._debug_dump("RX telemetry", status, raw=data)
+        if len(status.inverters) == 0:
+            return
+
+        self._last_status = status
+        self._last_telemetry_time = time.time()
+        self.update_success_timestamp()
+
+        try:
+            plan, result = await asyncio.to_thread(
+                runtime.ingest_gateway_status,
+                status,
+                prefix=self.prefix,
+                serial_filter=tuple(self.gateway_inverter_serial),
+            )
+            if plan is None or not getattr(result, "accepted", False):
+                raise ValueError("Lattice runtime did not accept the Gateway status")
+            self._synchronize_lattice_gateway_plan(plan)
+            self._inject_entities(status)
+            self._lattice_gateway_stale = False
+        except Exception as e:
+            self._auto_configured = False
+            self.log(f"Warn: GatewayMQTT: Lattice auto-config rejected Gateway telemetry: {e}")
+            await self._set_lattice_gateway_liveness(runtime, False)
+            return
+
+        if not self.api_started:
+            self.api_started = True
+            self.log("Info: GatewayMQTT: First telemetry published to Lattice and auto-config complete, API started")
+
     def _debug_dump(self, label, message=None, raw=None, message_type=None):
         """Log a protobuf message as readable text when debug logging is enabled.
 

apps/predbat/gecloud.py

@@ -1,3 +1,4 @@
+# cspell:ignore autoconfig
 # -----------------------------------------------------------------------------
 # Predbat Home Battery System
 # Copyright Trefor Southwell 2026 - All Rights Reserved
@@ -821,6 +822,65 @@ async def publish_registers(self, device, registers, select_key=None):
                 self.dashboard_item(entity_id, state="on" if state else "off", attributes=attributes, app="gecloud")
                 self.register_entity_map[entity_id] = {"device": device, "key": key}
 
+    def _lattice_autoconfig_runtime(self):
+        """Return the enabled runtime only for GE automatic configuration."""
+        if not self.automatic:
+            return None
+        runtime = getattr(self.base, "lattice_autoconfig_runtime", None)
+        if runtime is None or getattr(runtime, "enabled", False) is not True:
+            return None
+        if runtime.provider_active("ge-cloud") is not True:
+            return None
+        return runtime
+
+    async def stop(self):
+        """Invalidate the long-lived provider before a component restart."""
+        runtime = self._lattice_autoconfig_runtime()
+        if runtime is not None:
+            await self._set_lattice_liveness(False)
+        await super().stop()
+
+    async def _refresh_automatic_config(self):
+        """Select exactly one legacy or Lattice automatic-config writer."""
+        if not self.automatic:
+            return False
+        runtime = self._lattice_autoconfig_runtime()
+        if runtime is None:
+            await self.async_automatic_config(self.devices_dict)
+            return True
+        await asyncio.to_thread(
+            runtime.ingest_gecloud_state,
+            self.devices_dict,
+            self.settings,
+            self.info,
+            prefix=self.prefix,
+            load_today_ignore=self.get_arg(
+                "ge_cloud_load_today_ignore",
+                default=False,
+            ),
+            split_pv=self.get_arg(
+                "ge_cloud_automatic_split_pv",
+                default=False,
+            ),
+            split_ct=self.get_arg(
+                "ge_cloud_automatic_split_ct",
+                default=False,
+            ),
+            shared_ct=self.get_arg(
+                "ge_cloud_automatic_shared_ct",
+                default=False,
+            ),
+        )
+        return True
+
+    async def _set_lattice_liveness(self, health):
+        """Publish one bounded cloud-health transition when Lattice is active."""
+        runtime = self._lattice_autoconfig_runtime()
+        if runtime is None:
+            return False
+        await asyncio.to_thread(runtime.set_gecloud_liveness, health)
+        return True
+
     async def async_automatic_config(self, devices):
         """
         Automatically configure predbat using GE Cloud auto-detected devices.
@@ -1109,7 +1169,12 @@ async def run(self, seconds, first):
                 else:
                     self.log("GECloud: No valid settings found in storage cache, will poll")
 
+        lattice_runtime = self._lattice_autoconfig_runtime()
+        lattice_refresh_due = False
+        lattice_health = "unchanged"
+
         if first or (seconds % 120 == 0):
+            lattice_success_before = self.last_success_timestamp if lattice_runtime is not None else None
             inverter_auth_denied = False
             for device in self.device_list:
                 self.status[device] = await self.async_get_inverter_status(device, self.status.get(device, {}))
@@ -1123,6 +1188,15 @@ async def run(self, seconds, first):
                 self.info[device] = await self.async_get_device_info(device, self.info.get(device, {}))
                 await self.publish_info(device, self.info[device])
 
+            if lattice_runtime is not None:
+                if inverter_auth_denied:
+                    lattice_health = False
+                elif self.last_success_timestamp != lattice_success_before:
+                    lattice_health = True
+                    lattice_refresh_due = True
+                else:
+                    lattice_health = None
+
             # Surface a clear, correct status when the GivEnergy cloud API denied access to the core
             # inverter data, rather than letting stale data be misdiagnosed downstream (e.g. as
             # inverter clock skew). Scoped to inverter — not EVC — auth failures, and reported only
@@ -1163,7 +1237,9 @@ async def run(self, seconds, first):
             # One shot tasks
             if first:
                 if self.automatic:
-                    await self.async_automatic_config(self.devices_dict)
+                    if lattice_runtime is None or lattice_health is True:
+                        await self._refresh_automatic_config()
+                    lattice_refresh_due = False
 
             now_utc = self.now_utc_exact
             options_due = self.default_options_stamp is None or (now_utc - self.default_options_stamp) >= timedelta(hours=24)
@@ -1172,6 +1248,12 @@ async def run(self, seconds, first):
                 for device in self.device_list:
                     await self.enable_default_options(device, self.settings[device])
 
+        if lattice_runtime is not None:
+            if lattice_refresh_due and lattice_health is not False:
+                await self._refresh_automatic_config()
+            if lattice_health != "unchanged" and lattice_health is not True:
+                await self._set_lattice_liveness(lattice_health)
+
         # Clear pending writes
         for device in self.device_list:
             if device in self.pending_writes:

apps/predbat/lattice_autoconfig.py

@@ -1449,8 +1449,14 @@ def _field_provenance(
     return tuple(sorted(provenance, key=lambda item: (item.field_path, item.provider_id, item.generation, item.source_path)))
 
 
-def compile_auto_config(snapshots, user_overrides=()):
+def compile_auto_config(
+    snapshots,
+    user_overrides=(),
+    atomic_materializer=False,
+):
     """Compile usable provider snapshots into one deterministic immutable plan."""
+    if not isinstance(atomic_materializer, bool):
+        raise ValueError("atomic_materializer must be a boolean")
     snapshots = tuple(sorted(snapshots, key=lambda item: item.provider_id))
     if not snapshots:
         raise AutoConfigCompileError("no usable provider snapshots")
@@ -1516,9 +1522,11 @@ def compile_auto_config(snapshots, user_overrides=()):
         user_overrides,
     )
     if config_arguments:
-        blockers = tuple(blocker for blocker in readiness.blockers if blocker != "config_projection_bindings_missing") + ("atomic_materializer_missing",)
+        blockers = tuple(blocker for blocker in readiness.blockers if blocker != "config_projection_bindings_missing")
+        if not atomic_materializer:
+            blockers += ("atomic_materializer_missing",)
         readiness = MaterializationReadiness(
-            ready=False,
+            ready=not blockers,
             blockers=blockers,
         )
 
@@ -1666,10 +1674,22 @@ def compile_auto_config(snapshots, user_overrides=()):
 class LatticeAutoConfigCompiler:
     """Thread-safe invalidation, compilation, and last-known-good coordinator."""
 
-    def __init__(self, readers=None, materializer=None):
+    def __init__(
+        self,
+        readers=None,
+        materializer=None,
+        atomic_materializer=False,
+        allow_provider_failover=False,
+    ):
         """Create an idle compiler over provider snapshot readers."""
+        if not isinstance(atomic_materializer, bool):
+            raise ValueError("atomic_materializer must be a boolean")
+        if not isinstance(allow_provider_failover, bool):
+            raise ValueError("allow_provider_failover must be a boolean")
         self._readers = {}
         self._materializer = materializer
+        self._atomic_materializer = atomic_materializer
+        self._allow_provider_failover = allow_provider_failover
         self._lock = threading.RLock()
         self._compiling = False
         self._pending = False
@@ -1787,11 +1807,14 @@ def _compile_attempt(self):
             active_providers = set(dict(self._active_plan.provider_generations)) if self._active_plan is not None else set()
         usable_providers = {snapshot.provider_id for snapshot in snapshots}
         unavailable_active = sorted(active_providers - usable_providers)
-        if unavailable_active:
+        if unavailable_active and not self._allow_provider_failover:
             detail = "previously active provider(s) unavailable: {}".format(", ".join(unavailable_active))
             return None, issues + (CompileIssue("active_provider_unavailable", detail), CompileIssue("compile_failed", detail))
         try:
-            plan = compile_auto_config(snapshots)
+            plan = compile_auto_config(
+                snapshots,
+                atomic_materializer=self._atomic_materializer,
+            )
         except (AutoConfigCompileError, TopologyValidationError, TypeError, ValueError) as exc:
             return None, issues + (CompileIssue("compile_failed", str(exc)),)
         return plan, issues