From 8a1b3cf952ba680281b83db31c3db3795f79a94d Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 28 Jul 2026 09:15:09 +0100 Subject: [PATCH 1/2] feat(lattice): gate live schedule writers --- apps/predbat/config.py | 13 + apps/predbat/execute.py | 3 + apps/predbat/gateway.py | 4 + apps/predbat/lattice_schedule_authority.py | 7 +- apps/predbat/lattice_schedule_integration.py | 889 ++++++++++++++++ .../tests/test_lattice_schedule_authority.py | 1 + .../test_lattice_schedule_integration.py | 945 ++++++++++++++++++ 7 files changed, 1861 insertions(+), 1 deletion(-) create mode 100644 apps/predbat/lattice_schedule_integration.py create mode 100644 apps/predbat/tests/test_lattice_schedule_integration.py diff --git a/apps/predbat/config.py b/apps/predbat/config.py index a158d6e2d..6a668b10d 100644 --- a/apps/predbat/config.py +++ b/apps/predbat/config.py @@ -63,6 +63,19 @@ "default": False, "enable": "lattice_control_enable", }, + { + "name": "lattice_schedule_shadow_enable", + "friendly_name": "Lattice Schedule Shadow (experimental)", + "type": "switch", + "default": False, + }, + { + "name": "lattice_schedule_control_enable", + "friendly_name": "Lattice Schedule Control (experimental)", + "type": "switch", + "default": False, + "enable": "lattice_control_enable", + }, { "name": "active", "friendly_name": "Predbat Active", diff --git a/apps/predbat/execute.py b/apps/predbat/execute.py index 370517a0b..769c9e95c 100644 --- a/apps/predbat/execute.py +++ b/apps/predbat/execute.py @@ -20,6 +20,7 @@ from utils import dp0, dp2, dp3, calc_percent_limit, find_charge_rate from predbat_metrics import metrics from inverter import Inverter +from lattice_schedule_integration import lattice_balance_writer, lattice_plan_writer import time """ @@ -35,6 +36,7 @@ class Execute: adjustment, and multi-inverter balancing. """ + @lattice_plan_writer def execute_plan(self): status_extra = "" # extra status text added to Predbat notifications status_hold_car = "" # car hold status text @@ -945,6 +947,7 @@ def publish_inverter_data(self): }, ) + @lattice_balance_writer def balance_inverters(self, test_mode=False): """ Attempt to balance multiple inverters diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index 50be0ec90..44abfdebd 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -20,6 +20,7 @@ from component_base import ComponentBase from lattice_control import CAPABILITY_CHARGE_POWER_LIMIT, LatticeScalarDispatcher +from lattice_schedule_integration import lattice_async_writer from lattice_topology import LatticeTopologyStore, TopologyValidationError try: @@ -1574,6 +1575,7 @@ def _plan_changed(self, plan_entries): return True return plan_entries != self._last_published_plan + @lattice_async_writer("gateway legacy plan publish") async def publish_plan(self, plan_entries, timezone_str): """Build and publish an ExecutionPlan protobuf to the gateway. @@ -1604,6 +1606,7 @@ async def publish_plan(self, plan_entries, timezone_str): self._last_published_plan = plan_entries self.log(f"Info: GatewayMQTT: Published execution plan v{self._plan_version} ({len(plan_entries)} entries)") + @lattice_async_writer("gateway legacy plan republish") async def _republish_plan_if_stale(self): """Re-publish the last plan periodically so its embedded timestamp stays fresh. @@ -1625,6 +1628,7 @@ async def _republish_plan_if_stale(self): self._last_plan_publish_time = time.time() self.log("Info: GatewayMQTT: Re-published execution plan (refreshed timestamp)") + @lattice_async_writer(lambda _gateway, command, **_kwargs: "gateway legacy command {}".format(command)) async def publish_command(self, command, **kwargs): """Build and publish a JSON command to the gateway. diff --git a/apps/predbat/lattice_schedule_authority.py b/apps/predbat/lattice_schedule_authority.py index 581d301c2..7d40c0742 100644 --- a/apps/predbat/lattice_schedule_authority.py +++ b/apps/predbat/lattice_schedule_authority.py @@ -409,6 +409,8 @@ class AuthorityDecision: command_id: str = "" legacy_suppressed: bool = False ending_transition_required: bool = False + installed_plan_digest: str = "" + durable_state_present: bool = False fallback_allowed: bool = False quarantined: bool = False detail: str = "" @@ -1062,9 +1064,11 @@ def _decision_from_snapshot(snapshot, now_ts_ms): decision, legacy_suppressed=True, ending_transition_required=now_ts_ms >= installed["valid_until_ts_ms"], + installed_plan_digest=installed["applied_plan_digest"], + durable_state_present=record is not None, fallback_allowed=False, ) - return decision + return replace(decision, durable_state_present=record is not None) class LatticeScheduleAuthority: @@ -1339,6 +1343,7 @@ async def legacy_decision(self, now_ts_ms): return AuthorityDecision( RESULT_UNKNOWN, legacy_suppressed=True, + durable_state_present=snapshot is not _LOAD_FAILED, quarantined=True, detail="corrupt persisted authority state", ) diff --git a/apps/predbat/lattice_schedule_integration.py b/apps/predbat/lattice_schedule_integration.py new file mode 100644 index 000000000..4d5e6ed50 --- /dev/null +++ b/apps/predbat/lattice_schedule_integration.py @@ -0,0 +1,889 @@ +"""Default-off integration seam between PredBat plans and Lattice authority. + +This module deliberately owns no transport. An ACTIVE integration must inject +complete, topology-bound target, lease, send, and result-query bindings. Until +then ACTIVE is fail-closed, while OFF and SHADOW preserve legacy behaviour for +installations which have never armed Lattice schedule control. + +ACTIVE must remain disabled until the gateway runtime atomically fences or +retires any retained legacy execution plan before accepting Lattice authority. +This client-side barrier prevents new overlapping writers but cannot prove that +device-side prerequisite by itself. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import inspect +import json +import threading +from dataclasses import dataclass, field, replace +from datetime import datetime, timezone +from functools import wraps +from typing import Any, Callable + +from const import MINUTE_WATT +from ha import run_async +from lattice_schedule_authority import ( + CompletePlan, + EXECUTION_NATIVE, + LatticeScheduleAuthority, + PlanWindow, + RESULT_APPLIED, + RESULT_NOT_APPLIED, + RESULT_UNKNOWN, + ScheduleLease, + ScheduleTarget, + ScheduleValidationError, + compile_absolute_schedule, +) +from utils import calc_percent_limit + + +MODE_OFF = "OFF" +MODE_SHADOW = "SHADOW" +MODE_ACTIVE = "ACTIVE" + +_MARKER_MODULE = "lattice_schedule_integration" +_MARKER_KEY = "ever_armed" +_MARKER_SCHEMA = 1 +_COORDINATOR_LOCK = threading.Lock() + + +@dataclass(frozen=True) +class ScheduleRuntimeBindings: + """Complete runtime dependencies required before ACTIVE may send.""" + + scope_id: str + plan_builder: Callable[[Any], Any] + resolve_target: Callable[[Any], Any] + resolve_cancellation_target: Callable[[str, str], Any] + acquire_lease: Callable[[Any, int], Any] + send: Callable[[Any], Any] + query_result: Callable[[str], Any] + storage: Any = None + + @property + def complete(self): + """Return whether every safety-critical binding is present.""" + return ( + isinstance(self.scope_id, str) + and bool(self.scope_id) + and all( + callable(value) + for value in ( + self.plan_builder, + self.resolve_target, + self.resolve_cancellation_target, + self.acquire_lease, + self.send, + self.query_result, + ) + ) + ) + + +@dataclass(frozen=True) +class ScheduleIntegrationDecision: + """One shared decision for all legacy battery-control writers.""" + + allow_legacy: bool + mode: str + state: str = RESULT_NOT_APPLIED + command_id: str = "" + detail: str = "" + shadow_digest: str = "" + _writer_permit: Any = field(default=None, repr=False, compare=False) + + +class LegacyWriterPermit: + """One safely releasable admission held for a complete legacy write.""" + + def __init__(self, gate): + self._gate = gate + self._released = False + self._lock = threading.Lock() + + def release(self): + """Release this writer exactly once.""" + with self._lock: + if self._released: + return + self._released = True + self._gate.release_writer() + + +class LegacyWriteGate: + """Process-wide transition barrier and admitted-writer accounting.""" + + def __init__(self): + self._condition = threading.Condition() + self._allowed = True + self._reason = "" + self._transition = None + self._active_writers = 0 + + def begin_transition(self, reason): + """Close admission, serialize decisions, and drain admitted writers.""" + token = object() + with self._condition: + while self._transition is not None: + self._condition.wait() + self._transition = token + self._allowed = False + self._reason = str(reason) + while self._active_writers: + self._condition.wait() + return token + + def finish_transition(self, token, allow_legacy, reason="", acquire_writer=False): + """Atomically publish a decision and optionally admit its first writer.""" + with self._condition: + if self._transition is not token: + raise RuntimeError("legacy writer transition token is not current") + permit = None + self._allowed = bool(allow_legacy) + self._reason = "" if allow_legacy else str(reason) + if allow_legacy and acquire_writer: + self._active_writers += 1 + permit = LegacyWriterPermit(self) + self._transition = None + self._condition.notify_all() + return permit + + def release_writer(self): + """Release one admitted writer and wake an exclusive transition.""" + with self._condition: + if self._active_writers <= 0: + raise RuntimeError("legacy writer permit underflow") + self._active_writers -= 1 + if not self._active_writers: + self._condition.notify_all() + + @property + def allowed(self): + """Return the current atomic writer permit.""" + with self._condition: + return self._allowed + + @property + def reason(self): + """Return the diagnostic reason for a closed gate.""" + with self._condition: + return self._reason + + @property + def active_writers(self): + """Return the number of complete legacy operations still admitted.""" + with self._condition: + return self._active_writers + + +def _flag(value): + """Interpret only explicit boolean-like values as enabled.""" + if value is True: + return True + if isinstance(value, str): + return value.strip().lower() in ("1", "true", "yes", "on", "enable", "enabled") + return False + + +def _owner(subject): + """Return the main PredBat object for a mixin or component.""" + base = getattr(subject, "base", None) + if base is not None and callable(getattr(base, "get_arg", None)): + return base + return subject + + +def _get_arg(base, name, default=False): + """Read one option without allowing test doubles to become truthy flags.""" + getter = getattr(base, "get_arg", None) + if not callable(getter): + return default + try: + return getter(name, default) + except Exception: + return default + + +def _now_utc(base): + """Return PredBat's aware UTC clock, rejecting missing or invalid values.""" + value = getattr(base, "now_utc", None) + if not isinstance(value, datetime) or value.tzinfo is None or value.utcoffset() is None: + raise ScheduleValidationError("now_utc must be timezone-aware") + return value.astimezone(timezone.utc) + + +def _invoke(callback, *args): + """Invoke one injected binding and await it when necessary.""" + value = callback(*args) + if inspect.isawaitable(value): + return value + + async def completed(): + return value + + return completed() + + +def _storage_from(base, bindings=None): + """Resolve only an async storage implementation, never an arbitrary mock.""" + if bindings is not None and bindings.storage is not None: + candidate = bindings.storage + else: + components = getattr(base, "components", None) + getter = getattr(components, "get_component", None) + if not callable(getter): + return None + try: + candidate = getter("storage") + except Exception: + return None + if candidate is None: + return None + if not inspect.iscoroutinefunction(getattr(candidate, "load", None)): + return None + if not inspect.iscoroutinefunction(getattr(candidate, "save", None)): + return None + return candidate + + +def _window_bounds(window): + """Validate and return one optimizer window's integer minute bounds.""" + if not isinstance(window, dict): + raise ScheduleValidationError("optimizer window must be a mapping") + start = window.get("start") + end = window.get("end") + if isinstance(start, bool) or not isinstance(start, int) or isinstance(end, bool) or not isinstance(end, int): + raise ScheduleValidationError("optimizer window bounds must be integers") + if start < 0 or end <= start: + raise ScheduleValidationError("optimizer window must be non-empty") + return start, end + + +def build_complete_plan(base, altitude_id="predbat-battery"): + """Normalize PredBat's complete best plan for the pure v0.4 compiler.""" + now_utc = _now_utc(base) + timeline_start = getattr(base, "midnight_utc", None) + if not isinstance(timeline_start, datetime) or timeline_start.tzinfo is None or timeline_start.utcoffset() is None: + timeline_start = now_utc.replace(hour=0, minute=0, second=0, microsecond=0) + timeline_start = timeline_start.astimezone(timezone.utc) + + charge_windows = tuple(getattr(base, "charge_window_best", ()) or ()) + charge_limits = tuple(getattr(base, "charge_limit_best", ()) or ()) + export_windows = tuple(getattr(base, "export_window_best", ()) or ()) + export_limits = tuple(getattr(base, "export_limits_best", ()) or ()) + if len(charge_windows) != len(charge_limits) or len(export_windows) != len(export_limits): + raise ScheduleValidationError("optimizer plan windows and limits are incomplete") + + soc_max = getattr(base, "soc_max", 0) + if not isinstance(soc_max, (int, float)) or isinstance(soc_max, bool) or soc_max <= 0: + raise ScheduleValidationError("optimizer battery capacity is invalid") + reserve = getattr(base, "reserve", 0) + reserve_percent = calc_percent_limit(reserve, soc_max) + charge_power = float(getattr(base, "battery_rate_max_charge", 0) * MINUTE_WATT) + discharge_power = float(getattr(base, "battery_rate_max_export", getattr(base, "battery_rate_max_discharge", 0)) * MINUTE_WATT) + + windows = [] + latest_end = 0 + for index, window in enumerate(charge_windows): + start, end = _window_bounds(window) + latest_end = max(latest_end, end) + target_soc = calc_percent_limit(charge_limits[index], soc_max) + if target_soc == reserve_percent: + windows.append( + PlanWindow( + altitude_id, + start, + end, + "hold", + target_soc=float(target_soc), + reserve_soc=float(reserve_percent), + charge_power_limit=0.0, + enable=True, + ) + ) + else: + windows.append( + PlanWindow( + altitude_id, + start, + end, + "force_charge", + target_soc=float(target_soc), + charge_power_limit=charge_power, + enable=True, + ) + ) + + for index, window in enumerate(export_windows): + start, end = _window_bounds(window) + latest_end = max(latest_end, end) + target_soc = export_limits[index] + if isinstance(target_soc, bool) or not isinstance(target_soc, (int, float)): + raise ScheduleValidationError("optimizer export limit is invalid") + windows.append( + PlanWindow( + altitude_id, + start, + end, + "force_export", + target_soc=float(target_soc), + discharge_power_limit=discharge_power, + enable=True, + ) + ) + + configured_horizon = getattr(base, "forecast_plan_hours", 8) + if isinstance(configured_horizon, bool) or not isinstance(configured_horizon, (int, float)) or configured_horizon <= 0: + raise ScheduleValidationError("optimizer plan horizon is invalid") + now_offset = int((now_utc - timeline_start).total_seconds() // 60) + horizon_minutes = max(latest_end, now_offset + int(configured_horizon * 60), now_offset + 1) + timezone_name = str(getattr(base, "local_tz", "UTC")) + return CompletePlan( + altitude_id=altitude_id, + timeline_start_utc=timeline_start, + horizon_minutes=horizon_minutes, + diagnostic_timezone=timezone_name, + windows=tuple(windows), + default_mode="self_use", + execution=EXECUTION_NATIVE, + is_complete=True, + ) + + +def stable_command_id(schedule, target, lease): + """Bind replay identity to the exact schedule, target, and admission lease.""" + value = { + "plan_digest": schedule.plan_digest, + "target": { + "provider": target.provider, + "node_id": target.node_id, + "access_path": target.access_path, + "topology_version": target.topology_version, + "doc_version": target.doc_version, + "cap_ref": target.cap_ref, + "altitude_id": target.altitude_id, + "scope_id": target.scope_id, + "offer": target.offer.fingerprint, + "owned_node_ids": list(target.owned_node_ids), + }, + "lease": { + "origin_id": lease.origin_id, + "controller_class": lease.controller_class, + "lease_id": lease.lease_id, + "scope_id": lease.scope_id, + "fencing_token": lease.fencing_token, + "expires_ts_ms": lease.expires_ts_ms, + }, + } + encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + return "predbat-" + hashlib.sha256(encoded).hexdigest()[:48] + + +def stable_cancellation_command_id(expected_plan_digest, reason, target, lease): + """Bind one ending transition to exact installed authority and admission.""" + value = { + "transition": "CANCEL", + "expected_plan_digest": expected_plan_digest, + "reason": reason, + "target": { + "provider": target.provider, + "node_id": target.node_id, + "access_path": target.access_path, + "topology_version": target.topology_version, + "doc_version": target.doc_version, + "cap_ref": target.cap_ref, + "altitude_id": target.altitude_id, + "scope_id": target.scope_id, + "offer": target.offer.fingerprint, + "owned_node_ids": list(target.owned_node_ids), + }, + "lease": { + "origin_id": lease.origin_id, + "controller_class": lease.controller_class, + "lease_id": lease.lease_id, + "scope_id": lease.scope_id, + "fencing_token": lease.fencing_token, + "expires_ts_ms": lease.expires_ts_ms, + }, + } + encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + return "predbat-cancel-" + hashlib.sha256(encoded).hexdigest()[:41] + + +class LatticeScheduleCoordinator: + """Coordinate one shared legacy/Lattice writer decision for PredBat.""" + + def __init__(self, base): + self.base = base + self.gate = LegacyWriteGate() + self._marker_loaded = False + self._marker = None + self._marker_corrupt = False + self._marker_load_failed = False + self._authority = None + self._authority_scope = "" + self._authority_binding_key = None + self.last_decision = ScheduleIntegrationDecision(True, MODE_OFF) + + def _mode(self): + shadow = _flag(_get_arg(self.base, "lattice_schedule_shadow_enable", False)) + active = _flag(_get_arg(self.base, "lattice_control_enable", False)) and _flag(_get_arg(self.base, "lattice_schedule_control_enable", False)) + if active: + return MODE_ACTIVE + if shadow: + return MODE_SHADOW + return MODE_OFF + + def _bindings(self): + value = getattr(self.base, "_lattice_schedule_bindings", None) + return value if isinstance(value, ScheduleRuntimeBindings) else None + + def _storage(self): + return _storage_from(self.base, self._bindings()) + + async def _load_marker(self): + if self._marker_loaded: + return + storage = self._storage() + if storage is None: + return + try: + marker = await storage.load(_MARKER_MODULE, _MARKER_KEY) + except Exception: + self._marker_load_failed = True + marker = None + if marker is None: + return + if not isinstance(marker, dict) or marker.get("schema_version") != _MARKER_SCHEMA or marker.get("ever_armed") is not True or not isinstance(marker.get("scope_id"), str) or not marker["scope_id"]: + self._marker_corrupt = True + else: + self._marker = dict(marker) + self._marker_loaded = True + + async def _arm(self, scope_id): + await self._load_marker() + if self._marker_corrupt or self._marker_load_failed: + return False + if self._marker is not None: + return self._marker["scope_id"] == scope_id + storage = self._storage() + if storage is None: + return False + marker = {"schema_version": _MARKER_SCHEMA, "ever_armed": True, "scope_id": scope_id} + try: + saved = bool(await storage.save(_MARKER_MODULE, _MARKER_KEY, marker, format="json", expiry=None)) + except Exception: + saved = False + if saved: + self._marker = marker + self._marker_loaded = True + return saved + + def _authority_for(self, scope_id, bindings=None): + storage = self._storage() + if storage is None: + return None + binding_key = ( + id(storage), + id(bindings.send) if bindings is not None else None, + id(bindings.query_result) if bindings is not None else None, + ) + if self._authority is not None and self._authority_scope == scope_id and self._authority_binding_key == binding_key: + return self._authority + + async def unavailable_send(_command): + return None + + async def unavailable_query(_command_id): + return None + + self._authority = LatticeScheduleAuthority( + storage=storage, + send=bindings.send if bindings is not None else unavailable_send, + query_result=bindings.query_result if bindings is not None else unavailable_query, + scope_id=scope_id, + ) + self._authority_scope = scope_id + self._authority_binding_key = binding_key + return self._authority + + def _record(self, decision): + self.last_decision = decision + return decision + + def _suppress(self, mode, detail, state=RESULT_NOT_APPLIED, command_id=""): + return self._record(ScheduleIntegrationDecision(False, mode, state, command_id, detail)) + + async def _request_ending_transition(self, mode, authority, authority_decision, bindings, now_ts_ms): + """Request one guarded cancellation without ever opening the legacy gate.""" + if mode != MODE_ACTIVE: + return self._suppress( + mode, + "installed Lattice schedule requires an explicit ending transition", + authority_decision.state, + authority_decision.command_id, + ) + if bindings is None or not bindings.complete: + return self._suppress( + mode, + "complete Lattice schedule runtime bindings are unavailable for ending transition", + authority_decision.state, + authority_decision.command_id, + ) + if not authority_decision.installed_plan_digest or bindings.scope_id != self._marker["scope_id"]: + return self._suppress( + mode, + "installed Lattice authority cannot be bound to a safe ending transition", + authority_decision.state, + authority_decision.command_id, + ) + try: + target = await _invoke( + bindings.resolve_cancellation_target, + bindings.scope_id, + authority_decision.installed_plan_digest, + ) + lease = await _invoke(bindings.acquire_lease, target, now_ts_ms) + except Exception as exc: + return self._suppress( + mode, + "Lattice ending target or lease resolution failed: {}".format(exc), + authority_decision.state, + authority_decision.command_id, + ) + if not isinstance(target, ScheduleTarget) or not isinstance(lease, ScheduleLease): + return self._suppress( + mode, + "Lattice ending target or lease resolution returned an invalid type", + authority_decision.state, + authority_decision.command_id, + ) + if target.scope_id != bindings.scope_id or lease.scope_id != bindings.scope_id: + return self._suppress( + mode, + "Lattice ending target, lease, and configured scope differ", + authority_decision.state, + authority_decision.command_id, + ) + command_id = stable_cancellation_command_id( + authority_decision.installed_plan_digest, + "VALIDITY_END", + target, + lease, + ) + ending = await authority.cancel( + target, + lease, + authority_decision.installed_plan_digest, + "VALIDITY_END", + now_ts_ms, + command_id, + ) + if ending.state == RESULT_APPLIED and not ending.legacy_suppressed: + return None + return self._suppress( + mode, + ending.detail or "Lattice ending transition has not proven authority release", + ending.state, + ending.command_id, + ) + + async def _prior_schedule_decision(self, mode, now_ts_ms): + await self._load_marker() + if self._marker_load_failed: + return self._suppress(mode, "durable Lattice activation marker could not be loaded") + if self._marker_corrupt: + return self._suppress(mode, "durable Lattice activation marker is corrupt") + if self._marker is None: + return None + authority = self._authority_for(self._marker["scope_id"], self._bindings()) + if authority is None: + return self._suppress(mode, "durable Lattice authority storage is unavailable") + decision = await authority.legacy_decision(now_ts_ms) + if not decision.durable_state_present: + return self._suppress( + mode, + "durable Lattice authority state is missing after activation", + RESULT_UNKNOWN, + decision.command_id, + ) + if decision.ending_transition_required: + return await self._request_ending_transition( + mode, + authority, + decision, + self._bindings(), + now_ts_ms, + ) + if decision.legacy_suppressed or decision.quarantined: + return self._suppress(mode, decision.detail or "durable Lattice schedule owns this scope", decision.state, decision.command_id) + return None + + async def _evaluate_before_plan(self, mode): + """Evaluate plan authority while the exclusive transition is held.""" + await self._load_marker() + if self._marker_load_failed: + return self._suppress(mode, "durable Lattice activation marker could not be loaded") + if self._marker_corrupt: + return self._suppress(mode, "durable Lattice activation marker is corrupt") + if self._marker is not None and self._storage() is None: + return self._suppress(mode, "durable Lattice authority storage is unavailable") + + try: + now_utc = _now_utc(self.base) + now_ts_ms = int(now_utc.timestamp() * 1000) + except Exception as exc: + if mode == MODE_ACTIVE or self._marker is not None: + return self._suppress(mode, str(exc)) + return self._record(ScheduleIntegrationDecision(True, mode, detail=str(exc))) + + prior = await self._prior_schedule_decision(mode, now_ts_ms) + if prior is not None: + if mode != MODE_ACTIVE or prior.state != RESULT_APPLIED: + return prior + # ACTIVE may atomically replace a still-valid installed schedule. + # The shared gate remains closed throughout compilation and dispatch. + + if mode == MODE_OFF: + return self._record(ScheduleIntegrationDecision(True, mode)) + + if mode == MODE_SHADOW: + try: + complete_plan = build_complete_plan(self.base) + schedule = compile_absolute_schedule(complete_plan, now_utc) + except Exception as exc: + return self._record(ScheduleIntegrationDecision(True, mode, detail=str(exc))) + return self._record(ScheduleIntegrationDecision(True, mode, shadow_digest=schedule.plan_digest)) + + bindings = self._bindings() + if bindings is None or not bindings.complete: + return self._suppress(mode, "complete Lattice schedule runtime bindings are unavailable") + if self._storage() is None: + return self._suppress(mode, "durable Lattice authority storage is unavailable") + + try: + complete_plan = await _invoke(bindings.plan_builder, self.base) + if not isinstance(complete_plan, CompletePlan): + raise ScheduleValidationError("injected ACTIVE plan builder did not return CompletePlan") + schedule = compile_absolute_schedule(complete_plan, now_utc) + except Exception as exc: + return self._suppress(mode, str(exc)) + + try: + target = await _invoke(bindings.resolve_target, schedule) + lease = await _invoke(bindings.acquire_lease, target, now_ts_ms) + except Exception as exc: + return self._suppress(mode, "Lattice target or lease resolution failed: {}".format(exc)) + if not isinstance(target, ScheduleTarget) or not isinstance(lease, ScheduleLease): + return self._suppress(mode, "Lattice target or lease resolution returned an invalid type") + if target.scope_id != bindings.scope_id or lease.scope_id != bindings.scope_id: + return self._suppress(mode, "Lattice target, lease, and configured scope differ") + if not await self._arm(bindings.scope_id): + return self._suppress(mode, "durable Lattice activation marker could not be persisted") + + authority = self._authority_for(bindings.scope_id, bindings) + if authority is None: + return self._suppress(mode, "durable Lattice authority storage is unavailable") + command_id = stable_command_id(schedule, target, lease) + authority_decision = await authority.dispatch(schedule, target, lease, now_ts_ms, command_id) + return self._suppress( + mode, + authority_decision.detail or "Lattice schedule authority suppresses legacy writers", + authority_decision.state, + authority_decision.command_id, + ) + + async def _begin_transition(self, reason): + """Enter the blocking writer barrier without blocking the event loop.""" + loop = asyncio.get_running_loop() + future = loop.run_in_executor(None, self.gate.begin_transition, reason) + try: + return await asyncio.shield(future) + except BaseException: + token = await asyncio.shield(future) + self.gate.finish_transition(token, False, "Lattice authority transition requester was cancelled") + raise + + def _finish_transition(self, token, decision, acquire_writer): + """Publish one decision and atomically attach its admitted writer.""" + permit = self.gate.finish_transition( + token, + decision.allow_legacy, + decision.detail, + acquire_writer=acquire_writer and decision.allow_legacy, + ) + if permit is not None: + return replace(decision, _writer_permit=permit) + return decision + + async def before_plan(self): + """Evaluate authority and atomically admit one complete plan writer.""" + token = await self._begin_transition("Lattice schedule authority is reconciling before plan execution") + try: + mode = self._mode() + decision = await self._evaluate_before_plan(mode) + except BaseException as exc: + self.gate.finish_transition(token, False, "Lattice schedule decision failed: {}".format(exc)) + raise + return self._finish_transition(token, decision, acquire_writer=True) + + async def _evaluate_legacy_writer(self, writer, mode): + """Refresh durable ownership while the exclusive transition is held.""" + await self._load_marker() + if self._marker_load_failed: + return self._suppress(mode, "durable Lattice activation marker could not be loaded") + if self._marker_corrupt: + return self._suppress(mode, "durable Lattice activation marker is corrupt") + if self._marker is not None and self._storage() is None: + return self._suppress(mode, "durable Lattice authority storage is unavailable") + try: + now_ts_ms = int(_now_utc(self.base).timestamp() * 1000) + except Exception as exc: + if mode == MODE_ACTIVE or self._marker is not None: + return self._suppress(mode, "{} clock is invalid: {}".format(writer, exc)) + return self._record(ScheduleIntegrationDecision(True, mode, detail=str(exc))) + prior = await self._prior_schedule_decision(mode, now_ts_ms) + if prior is not None: + return prior + if mode == MODE_ACTIVE: + return self._suppress(mode, "{} is blocked while Lattice schedule control is ACTIVE".format(writer)) + return self._record(ScheduleIntegrationDecision(True, mode)) + + async def acquire_legacy_writer(self, writer): + """Atomically decide and admit one complete non-plan legacy writer.""" + token = await self._begin_transition("{} is waiting for a Lattice authority decision".format(writer)) + try: + mode = self._mode() + decision = await self._evaluate_legacy_writer(writer, mode) + except BaseException as exc: + self.gate.finish_transition(token, False, "{} authority decision failed: {}".format(writer, exc)) + raise + decision = self._finish_transition(token, decision, acquire_writer=True) + return decision._writer_permit + + async def legacy_permitted(self, writer): + """Compatibility probe which never leaves an unowned writer permit.""" + permit = await self.acquire_legacy_writer(writer) + if permit is None: + return False + permit.release() + return True + + +def get_schedule_coordinator(subject): + """Return the one coordinator shared by PredBat and all components.""" + base = _owner(subject) + coordinator = getattr(base, "_lattice_schedule_coordinator", None) + if isinstance(coordinator, LatticeScheduleCoordinator): + return coordinator + with _COORDINATOR_LOCK: + coordinator = getattr(base, "_lattice_schedule_coordinator", None) + if not isinstance(coordinator, LatticeScheduleCoordinator): + coordinator = LatticeScheduleCoordinator(base) + setattr(base, "_lattice_schedule_coordinator", coordinator) + return coordinator + + +def before_legacy_plan(subject): + """Synchronous guard used at the first line of Execute.execute_plan.""" + coordinator = get_schedule_coordinator(subject) + return run_async(coordinator.before_plan()) + + +def acquire_legacy_writer(subject, writer): + """Synchronously acquire a permit held across one complete legacy writer.""" + coordinator = get_schedule_coordinator(subject) + return run_async(coordinator.acquire_legacy_writer(writer)) + + +async def acquire_legacy_writer_async(subject, writer): + """Asynchronously acquire a permit without blocking the caller's event loop.""" + coordinator = get_schedule_coordinator(subject) + return await coordinator.acquire_legacy_writer(writer) + + +def release_legacy_writer_permit(decision): + """Release the plan permit attached to an allowed integration decision.""" + permit = getattr(decision, "_writer_permit", None) + if permit is not None: + permit.release() + + +def legacy_write_permitted(subject, writer): + """Compatibility probe; complete writers must use a permit-holding wrapper.""" + coordinator = get_schedule_coordinator(subject) + return bool(run_async(coordinator.legacy_permitted(writer))) + + +async def legacy_write_permitted_async(subject, writer): + """Compatibility probe; complete writers must use a permit-holding wrapper.""" + coordinator = get_schedule_coordinator(subject) + return bool(await coordinator.legacy_permitted(writer)) + + +def lattice_plan_writer(method): + """Hold one legacy permit across complete Execute plan execution.""" + + @wraps(method) + def guarded(subject, *args, **kwargs): + decision = before_legacy_plan(subject) + if not decision.allow_legacy: + subject.log( + "Info: Lattice schedule authority suppressed legacy plan execution: state={} command_id={} detail={}".format( + decision.state, + decision.command_id, + decision.detail, + ) + ) + subject.set_charge_export_status(False, False, True) + subject.isCharging = False + subject.isExporting = False + return "Lattice schedule", decision.detail + try: + return method(subject, *args, **kwargs) + finally: + release_legacy_writer_permit(decision) + + return guarded + + +def lattice_balance_writer(method): + """Hold one legacy permit across complete inverter balancing.""" + + @wraps(method) + def guarded(subject, *args, **kwargs): + permit = acquire_legacy_writer(subject, "inverter balancing") + if permit is None: + subject.log("Info: Lattice schedule authority suppressed legacy inverter balancing") + return False + try: + return method(subject, *args, **kwargs) + finally: + permit.release() + + return guarded + + +def lattice_async_writer(writer): + """Decorate one async Gateway writer with complete permit lifetime.""" + + def decorate(method): + @wraps(method) + async def guarded(subject, *args, **kwargs): + label = writer(subject, *args, **kwargs) if callable(writer) else writer + permit = await acquire_legacy_writer_async(subject, label) + if permit is None: + subject.log("Info: GatewayMQTT: Lattice schedule authority suppressed {}".format(label)) + return None + try: + return await method(subject, *args, **kwargs) + finally: + permit.release() + + return guarded + + return decorate diff --git a/apps/predbat/tests/test_lattice_schedule_authority.py b/apps/predbat/tests/test_lattice_schedule_authority.py index 306a02f0d..44d3e63dc 100644 --- a/apps/predbat/tests/test_lattice_schedule_authority.py +++ b/apps/predbat/tests/test_lattice_schedule_authority.py @@ -345,6 +345,7 @@ def test_applied_suppresses_legacy_only_after_local_durable_save(self): self.assertFalse(decision.fallback_allowed) durable = asyncio.run(harness.authority.legacy_decision(NOW_MS)) self.assertTrue(durable.legacy_suppressed) + self.assertTrue(durable.durable_state_present) self.assertEqual(len(harness.sent), 1) def test_applied_receipt_save_failure_never_allows_fallback(self): diff --git a/apps/predbat/tests/test_lattice_schedule_integration.py b/apps/predbat/tests/test_lattice_schedule_integration.py new file mode 100644 index 000000000..87e742eed --- /dev/null +++ b/apps/predbat/tests/test_lattice_schedule_integration.py @@ -0,0 +1,945 @@ +"""Tests for the default-off live Lattice schedule integration seam.""" + +# cspell:words READBACK + +import asyncio +import copy +import threading +import time +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import predbat as _predbat # noqa: F401 - establish package import order before execute +from execute import Execute +from gateway import GatewayMQTT +from lattice_schedule_authority import ( + FALLBACK_SAFE, + RESULT_APPLIED, + RESULT_NOT_APPLIED, + RESULT_UNKNOWN, + VERIFICATION_DURABLE_STORE_READBACK, + ResolvedScheduleOffer, + ScheduleAck, + ScheduleLease, + ScheduleTarget, +) +from lattice_schedule_integration import ( + MODE_ACTIVE, + MODE_OFF, + MODE_SHADOW, + LatticeScheduleCoordinator, + ScheduleRuntimeBindings, + before_legacy_plan, + build_complete_plan, + get_schedule_coordinator, + lattice_async_writer, + lattice_balance_writer, + release_legacy_writer_permit, +) + + +NOW = datetime(2026, 7, 27, 12, 0, tzinfo=timezone.utc) +SCOPE = "hub:GW-1:battery" +ALTITUDE = "predbat-battery" + + +class FakeStorage: + """Asynchronous module/key storage shared across coordinator restarts.""" + + def __init__(self): + self.values = {} + self.save_count = 0 + self.fail_saves = False + self.fail_loads = False + + async def load(self, module, filename): + if self.fail_loads: + raise OSError("load failed") + return copy.deepcopy(self.values.get((module, filename))) + + async def save(self, module, filename, data, **_kwargs): + self.save_count += 1 + if self.fail_saves: + return False + self.values[(module, filename)] = copy.deepcopy(data) + return True + + +class FakeComponents: + """Expose only the storage component used by the seam.""" + + def __init__(self, storage): + self.storage = storage + + def get_component(self, name): + return self.storage if name == "storage" else None + + +class FakeBase: + """Minimal complete PredBat plan and configuration fixture.""" + + def __init__(self, storage=None): + self.options = { + "lattice_control_enable": False, + "lattice_schedule_control_enable": False, + "lattice_schedule_shadow_enable": False, + } + self.now_utc = NOW + self.midnight_utc = NOW.replace(hour=0, minute=0) + self.local_tz = timezone.utc + self.forecast_plan_hours = 8 + self.charge_window_best = [{"start": 13 * 60, "end": 14 * 60}] + self.charge_limit_best = [8.0] + self.export_window_best = [{"start": 17 * 60, "end": 18 * 60}] + self.export_limits_best = [20.0] + self.soc_max = 10.0 + self.reserve = 1.0 + self.battery_rate_max_charge = 50.0 + self.battery_rate_max_discharge = 50.0 + self.battery_rate_max_export = 50.0 + self.components = FakeComponents(storage) if storage is not None else None + self.logs = [] + self.status_calls = [] + + def get_arg(self, name, default=None, *args, **kwargs): + return self.options.get(name, default) + + def log(self, message): + self.logs.append(message) + + def set_charge_export_status(self, *args): + self.status_calls.append(args) + + +def resolved_target(): + """Return a topology-bound schedule target compatible with the fixture.""" + return ScheduleTarget( + provider="predbat-gateway", + node_id="GW-1", + access_path="gw-local", + topology_version="0.4.0", + doc_version=12, + cap_ref=41, + altitude_id=ALTITUDE, + scope_id=SCOPE, + offer=ResolvedScheduleOffer( + max_slots=8, + supported_modes=("self_use", "force_charge", "force_export", "hold"), + slot_fields=("target_soc", "reserve_soc", "charge_power_limit", "discharge_power_limit", "enable"), + execution_modes=("NATIVE",), + gap_policy="DEFAULT_MODE", + ), + owned_node_ids=("GW-1",), + ) + + +def lease(fence=10, expiry=None): + """Return one fresh AUTOMATION ownership lease.""" + return ScheduleLease( + origin_id="predbat", + controller_class="AUTOMATION", + lease_id="lease-{}".format(fence), + scope_id=SCOPE, + fencing_token=fence, + expires_ts_ms=int((expiry or (NOW + timedelta(minutes=30))).timestamp() * 1000), + ) + + +def applied(command): + """Return an exact durable APPLIED receipt.""" + return ScheduleAck( + command_id=command.command_id, + state=RESULT_APPLIED, + scope_id=SCOPE, + fencing_token=command.lease.fencing_token, + lease_id=command.lease.lease_id, + lease_expires_ts_ms=command.lease.expires_ts_ms, + origin_id=command.lease.origin_id, + controller_class=command.lease.controller_class, + plan_digest=command.schedule.plan_digest, + accepted_slot_count=len(command.schedule.slots), + durably_accepted=True, + valid_from_ts_ms=command.schedule.valid_from_ts_ms, + valid_until_ts_ms=command.schedule.valid_until_ts_ms, + verification=VERIFICATION_DURABLE_STORE_READBACK, + verified=True, + ) + + +def cancellation_applied(command, **overrides): + """Return exact durable proof that an ending transition released authority.""" + value = { + "command_id": command.command_id, + "state": RESULT_APPLIED, + "scope_id": SCOPE, + "fencing_token": command.lease.fencing_token, + "lease_id": command.lease.lease_id, + "lease_expires_ts_ms": command.lease.expires_ts_ms, + "origin_id": command.lease.origin_id, + "controller_class": command.lease.controller_class, + "cancelled_plan_digest": command.cancellation.expected_plan_digest, + "writer_exclusion_released": True, + "verification": VERIFICATION_DURABLE_STORE_READBACK, + "verified": True, + } + value.update(overrides) + return ScheduleAck(**value) + + +def rejected(command): + """Return one definite safe-to-fallback rejection.""" + return ScheduleAck( + command_id=command.command_id, + state=RESULT_NOT_APPLIED, + scope_id=SCOPE, + fencing_token=command.lease.fencing_token, + reason="UNSUPPORTED_OFFER", + fallback=FALLBACK_SAFE, + detail="schedule was not applied", + ) + + +def unknown(command): + """Return one ambiguous result which must keep all legacy writers closed.""" + return ScheduleAck( + command_id=command.command_id, + state=RESULT_UNKNOWN, + scope_id=SCOPE, + fencing_token=command.lease.fencing_token, + reason="EXECUTION_AMBIGUOUS", + fallback="BLOCKED", + detail="write completion is ambiguous", + ) + + +def install_bindings( + base, + storage, + send, + fence=10, + expiry=None, + query=None, + plan_builder=build_complete_plan, + cancellation_target=None, +): + """Inject all otherwise-unavailable runtime seams.""" + target = resolved_target() + current_lease = lease(fence, expiry) + base._lattice_schedule_bindings = ScheduleRuntimeBindings( + scope_id=SCOPE, + plan_builder=plan_builder, + resolve_target=lambda _schedule: target, + resolve_cancellation_target=(cancellation_target or (lambda _scope_id, _expected_plan_digest: target)), + acquire_lease=lambda _target, _now: current_lease, + send=send, + query_result=query or (lambda _command_id: None), + storage=storage, + ) + + +def activate(base): + """Enable both required ACTIVE gates.""" + base.options["lattice_control_enable"] = True + base.options["lattice_schedule_control_enable"] = True + + +def test_off_preserves_legacy_without_runtime_bindings(): + base = FakeBase() + decision = before_legacy_plan(base) + assert decision.allow_legacy is True + assert decision.mode == MODE_OFF + assert get_schedule_coordinator(base).gate.allowed is True + release_legacy_writer_permit(decision) + + +def test_shadow_compiles_digest_but_never_sends_or_persists_marker(): + storage = FakeStorage() + base = FakeBase(storage) + base.options["lattice_schedule_shadow_enable"] = True + sends = [] + install_bindings(base, storage, sends.append) + + decision = before_legacy_plan(base) + + assert decision.allow_legacy is True + assert decision.mode == MODE_SHADOW + assert len(decision.shadow_digest) == 64 + assert sends == [] + assert ("lattice_schedule_integration", "ever_armed") not in storage.values + assert ("lattice_schedule_authority", "schedule") not in storage.values + release_legacy_writer_permit(decision) + + +def test_active_without_complete_bindings_fails_closed_before_any_write(): + base = FakeBase() + activate(base) + + decision = before_legacy_plan(base) + + assert decision.allow_legacy is False + assert decision.mode == MODE_ACTIVE + assert "bindings are unavailable" in decision.detail + assert get_schedule_coordinator(base).gate.allowed is False + + +def test_safe_rejection_keeps_all_legacy_writers_closed_and_same_command_is_not_resent(): + storage = FakeStorage() + base = FakeBase(storage) + activate(base) + commands = [] + + def send(command): + commands.append(command) + return rejected(command) + + install_bindings(base, storage, send) + first = before_legacy_plan(base) + second = before_legacy_plan(base) + + assert first.allow_legacy is False + assert second.allow_legacy is False + assert first.state == RESULT_NOT_APPLIED + assert second.state == RESULT_NOT_APPLIED + assert first.command_id == second.command_id + assert len(commands) == 1 + assert storage.values[("lattice_schedule_integration", "ever_armed")]["ever_armed"] is True + assert asyncio.run(get_schedule_coordinator(base).legacy_permitted("gateway legacy command")) is False + assert get_schedule_coordinator(base).gate.allowed is False + + +def test_active_requires_an_injected_complete_plan_builder(): + storage = FakeStorage() + base = FakeBase(storage) + activate(base) + install_bindings(base, storage, rejected, plan_builder=None) + + decision = before_legacy_plan(base) + + assert decision.allow_legacy is False + assert "bindings are unavailable" in decision.detail + assert ("lattice_schedule_integration", "ever_armed") not in storage.values + + +def test_active_gate_is_closed_before_storage_plan_target_lease_and_send_work(): + events = [] + + class ObservedStorage(FakeStorage): + async def load(self, module, filename): + events.append(("storage-load", get_schedule_coordinator(base).gate.allowed)) + return await super().load(module, filename) + + async def save(self, module, filename, data, **kwargs): + events.append(("storage-save", get_schedule_coordinator(base).gate.allowed)) + return await super().save(module, filename, data, **kwargs) + + storage = ObservedStorage() + base = FakeBase(storage) + activate(base) + target = resolved_target() + current_lease = lease() + + def plan_builder(subject): + events.append(("plan-builder", get_schedule_coordinator(subject).gate.allowed)) + return build_complete_plan(subject) + + def resolve_target(_schedule): + events.append(("target", get_schedule_coordinator(base).gate.allowed)) + return target + + def acquire_lease(_target, _now): + events.append(("lease", get_schedule_coordinator(base).gate.allowed)) + return current_lease + + def send(command): + events.append(("send", get_schedule_coordinator(base).gate.allowed)) + return rejected(command) + + base._lattice_schedule_bindings = ScheduleRuntimeBindings( + scope_id=SCOPE, + plan_builder=plan_builder, + resolve_target=resolve_target, + resolve_cancellation_target=lambda _scope_id, _digest: target, + acquire_lease=acquire_lease, + send=send, + query_result=lambda _command_id: None, + storage=storage, + ) + + decision = before_legacy_plan(base) + + assert decision.allow_legacy is False + assert events + assert events[0][0] == "storage-load" + assert {"storage-load", "storage-save", "plan-builder", "target", "lease", "send"}.issubset({name for name, _allowed in events}) + assert all(allowed is False for _name, allowed in events) + + +def test_missing_clock_is_exact_legacy_off_shadow_diagnostic_and_active_closed(): + off = FakeBase() + del off.now_utc + off_decision = before_legacy_plan(off) + + shadow = FakeBase() + del shadow.now_utc + shadow.options["lattice_schedule_shadow_enable"] = True + shadow_decision = before_legacy_plan(shadow) + + active = FakeBase() + del active.now_utc + activate(active) + active_decision = before_legacy_plan(active) + + assert off_decision.allow_legacy is True + assert off_decision.mode == MODE_OFF + assert shadow_decision.allow_legacy is True + assert shadow_decision.mode == MODE_SHADOW + assert "timezone-aware" in shadow_decision.detail + assert active_decision.allow_legacy is False + assert active_decision.mode == MODE_ACTIVE + assert get_schedule_coordinator(active).gate.allowed is False + release_legacy_writer_permit(off_decision) + release_legacy_writer_permit(shadow_decision) + + +def test_storage_absence_is_not_cached_and_later_existing_marker_is_observed(): + storage = FakeStorage() + armed = FakeBase(storage) + activate(armed) + install_bindings(armed, storage, applied) + assert before_legacy_plan(armed).state == RESULT_APPLIED + + restarted = FakeBase() + first = before_legacy_plan(restarted) + release_legacy_writer_permit(first) + restarted.components = FakeComponents(storage) + second = before_legacy_plan(restarted) + + assert first.mode == MODE_OFF + assert first.allow_legacy is True + assert second.mode == MODE_OFF + assert second.allow_legacy is False + assert second.state == RESULT_APPLIED + + +def test_present_storage_marker_absence_is_not_cached_across_external_activation(): + """A later durable activation is observed even when storage existed at boot.""" + storage = FakeStorage() + restarted = FakeBase(storage) + first = before_legacy_plan(restarted) + assert first.allow_legacy is True + release_legacy_writer_permit(first) + + armed = FakeBase(storage) + activate(armed) + install_bindings(armed, storage, applied) + assert before_legacy_plan(armed).state == RESULT_APPLIED + + second = before_legacy_plan(restarted) + assert second.allow_legacy is False + assert second.state == RESULT_APPLIED + + +def test_armed_marker_without_authority_state_is_unknown_and_fail_closed(): + """Missing command state after activation is not inferred as non-application.""" + storage = FakeStorage() + storage.values[("lattice_schedule_integration", "ever_armed")] = { + "schema_version": 1, + "ever_armed": True, + "scope_id": SCOPE, + } + restarted = FakeBase(storage) + + decision = before_legacy_plan(restarted) + + assert decision.allow_legacy is False + assert decision.state == RESULT_UNKNOWN + assert "state is missing" in decision.detail + assert get_schedule_coordinator(restarted).gate.allowed is False + + +def test_invalid_restart_clock_loads_marker_before_non_plan_writer_decision(): + """Every non-plan writer stays closed when durable ownership cannot be timed.""" + storage = FakeStorage() + armed = FakeBase(storage) + activate(armed) + install_bindings(armed, storage, applied) + assert before_legacy_plan(armed).state == RESULT_APPLIED + + restarted = FakeBase(storage) + del restarted.now_utc + coordinator = get_schedule_coordinator(restarted) + + assert asyncio.run(coordinator.legacy_permitted("gateway legacy command")) is False + assert coordinator.gate.allowed is False + assert coordinator._marker["scope_id"] == SCOPE + + +def test_active_transition_drains_decorated_writer_before_dispatch(): + """Activation cannot overlap a complete legacy method already in flight.""" + storage = FakeStorage() + base = FakeBase(storage) + entered = threading.Event() + release = threading.Event() + completed = threading.Event() + commands = [] + + @lattice_balance_writer + def blocking_writer(subject): + entered.set() + assert release.wait(timeout=2) + completed.set() + return True + + writer_thread = threading.Thread(target=blocking_writer, args=(base,)) + writer_thread.start() + assert entered.wait(timeout=2) + assert get_schedule_coordinator(base).gate.active_writers == 1 + + activate(base) + install_bindings(base, storage, lambda command: commands.append(command) or rejected(command)) + activation_result = [] + activation_thread = threading.Thread(target=lambda: activation_result.append(before_legacy_plan(base))) + activation_thread.start() + + deadline = time.monotonic() + 2 + while get_schedule_coordinator(base).gate.allowed and time.monotonic() < deadline: + time.sleep(0.001) + assert get_schedule_coordinator(base).gate.allowed is False + assert activation_thread.is_alive() + assert commands == [] + + release.set() + writer_thread.join(timeout=2) + activation_thread.join(timeout=2) + + assert completed.is_set() + assert not writer_thread.is_alive() + assert not activation_thread.is_alive() + assert len(commands) == 1 + assert activation_result[0].allow_legacy is False + assert get_schedule_coordinator(base).gate.active_writers == 0 + + +def test_queued_writer_samples_active_mode_inside_exclusive_transition(): + """A writer queued while OFF cannot retain that stale mode across activation.""" + base = FakeBase() + coordinator = get_schedule_coordinator(base) + first = asyncio.run(coordinator.acquire_legacy_writer("first writer")) + assert first is not None + + queued = [] + queued_thread = threading.Thread(target=lambda: queued.append(asyncio.run(coordinator.acquire_legacy_writer("queued writer")))) + queued_thread.start() + deadline = time.monotonic() + 2 + while coordinator.gate.allowed and time.monotonic() < deadline: + time.sleep(0.001) + assert coordinator.gate.allowed is False + + activate(base) + first.release() + queued_thread.join(timeout=2) + + assert not queued_thread.is_alive() + assert queued == [None] + assert coordinator.gate.active_writers == 0 + assert coordinator.gate.allowed is False + + +def test_decorated_writer_exception_releases_permit(): + """A failed legacy method cannot strand ACTIVE behind an admitted writer.""" + base = FakeBase() + + @lattice_balance_writer + def broken_writer(_subject): + raise RuntimeError("write failed") + + try: + broken_writer(base) + except RuntimeError as exc: + assert str(exc) == "write failed" + else: + raise AssertionError("legacy writer exception was swallowed") + + assert get_schedule_coordinator(base).gate.active_writers == 0 + + +def test_applied_schedule_survives_mode_downgrade_restart_and_lease_expiry(): + storage = FakeStorage() + active = FakeBase(storage) + activate(active) + commands = [] + + def send(command): + commands.append(command) + return applied(command) + + install_bindings(active, storage, send, expiry=NOW + timedelta(minutes=15)) + accepted = before_legacy_plan(active) + assert accepted.allow_legacy is False + assert accepted.state == RESULT_APPLIED + + restarted = FakeBase(storage) + restarted.now_utc = NOW + timedelta(minutes=30) + restarted.midnight_utc = restarted.now_utc.replace(hour=0, minute=0) + downgraded = before_legacy_plan(restarted) + + assert downgraded.mode == MODE_OFF + assert downgraded.allow_legacy is False + assert downgraded.state == RESULT_APPLIED + assert "validity has expired" not in downgraded.detail + + +def test_shadow_still_suppresses_a_durable_prior_applied_schedule(): + storage = FakeStorage() + active = FakeBase(storage) + activate(active) + install_bindings(active, storage, applied) + assert before_legacy_plan(active).allow_legacy is False + + shadow = FakeBase(storage) + shadow.options["lattice_schedule_shadow_enable"] = True + decision = before_legacy_plan(shadow) + + assert decision.mode == MODE_SHADOW + assert decision.allow_legacy is False + assert decision.state == RESULT_APPLIED + + +def test_valid_until_reopens_legacy_after_restart(): + storage = FakeStorage() + active = FakeBase(storage) + activate(active) + install_bindings(active, storage, applied) + assert before_legacy_plan(active).allow_legacy is False + + expired = FakeBase(storage) + expired.now_utc = NOW.replace(hour=20) + expired.midnight_utc = expired.now_utc.replace(hour=0, minute=0) + sends = [] + install_bindings( + expired, + storage, + sends.append, + fence=11, + expiry=expired.now_utc + timedelta(minutes=30), + ) + decision = before_legacy_plan(expired) + + assert decision.mode == MODE_OFF + assert decision.allow_legacy is False + assert "ending transition" in decision.detail + assert sends == [] + + +def test_active_expiry_cancels_with_verified_release_before_replacement(): + """ACTIVE keeps the gate shut across guarded cancellation and replacement.""" + storage = FakeStorage() + initial = FakeBase(storage) + activate(initial) + install_bindings(initial, storage, applied) + assert before_legacy_plan(initial).state == RESULT_APPLIED + installed_digest = storage.values[("lattice_schedule_authority", "schedule")]["installed"]["applied_plan_digest"] + + expired = FakeBase(storage) + expired.now_utc = NOW.replace(hour=20) + expired.midnight_utc = expired.now_utc.replace(hour=0, minute=0) + activate(expired) + commands = [] + gate_states = [] + ending_requests = [] + + def cancellation_target(scope_id, expected_plan_digest): + ending_requests.append((scope_id, expected_plan_digest)) + return resolved_target() + + def send(command): + commands.append(command) + gate_states.append(get_schedule_coordinator(expired).gate.allowed) + if hasattr(command, "cancellation"): + return cancellation_applied(command) + return applied(command) + + install_bindings( + expired, + storage, + send, + fence=11, + expiry=expired.now_utc + timedelta(minutes=30), + cancellation_target=cancellation_target, + ) + decision = before_legacy_plan(expired) + + assert decision.state == RESULT_APPLIED + assert decision.allow_legacy is False + assert gate_states == [False, False] + assert ending_requests == [(SCOPE, installed_digest)] + assert len(commands) == 2 + assert "cancellation" in commands[0].to_wire()["schedule_transition"] + assert commands[0].cancellation.expected_plan_digest == installed_digest + assert "replacement" in commands[1].to_wire()["schedule_transition"] + durable = storage.values[("lattice_schedule_authority", "schedule")] + assert durable["installed"]["command_id"] == commands[1].command_id + + +def test_unproven_applied_cancellation_stays_unknown_and_never_replaces(): + """Bare, mismatched, or acceptance-only APPLIED cannot reopen either writer.""" + mutations = [ + {"cancelled_plan_digest": "f" * 64}, + {"writer_exclusion_released": False}, + { + "verification": "ACCEPTED_ONLY", + "verified": False, + }, + ] + for mutation in mutations: + storage = FakeStorage() + initial = FakeBase(storage) + activate(initial) + install_bindings(initial, storage, applied) + assert before_legacy_plan(initial).state == RESULT_APPLIED + installed_digest = storage.values[("lattice_schedule_authority", "schedule")]["installed"]["applied_plan_digest"] + + expired = FakeBase(storage) + expired.now_utc = NOW.replace(hour=20) + expired.midnight_utc = expired.now_utc.replace(hour=0, minute=0) + activate(expired) + commands = [] + + def send(command): + commands.append(command) + return cancellation_applied(command, **mutation) + + install_bindings( + expired, + storage, + send, + fence=11, + expiry=expired.now_utc + timedelta(minutes=30), + ) + decision = before_legacy_plan(expired) + + assert decision.state == RESULT_UNKNOWN + assert decision.allow_legacy is False + assert get_schedule_coordinator(expired).gate.allowed is False + assert len(commands) == 1 + durable = storage.values[("lattice_schedule_authority", "schedule")] + assert durable["installed"]["applied_plan_digest"] == installed_digest + + +def test_restart_reconciles_same_cancellation_id_without_republishing(): + """Power loss queries the ending command, then sends only the new schedule.""" + storage = FakeStorage() + initial = FakeBase(storage) + activate(initial) + install_bindings(initial, storage, applied) + assert before_legacy_plan(initial).state == RESULT_APPLIED + + expiry = NOW.replace(hour=20, minute=30) + interrupted = FakeBase(storage) + interrupted.now_utc = NOW.replace(hour=20) + interrupted.midnight_utc = interrupted.now_utc.replace(hour=0, minute=0) + activate(interrupted) + cancellation_commands = [] + + def lose_cancellation_result(command): + cancellation_commands.append(command) + return None + + install_bindings( + interrupted, + storage, + lose_cancellation_result, + fence=11, + expiry=expiry, + query=lambda _command_id: None, + ) + unresolved = before_legacy_plan(interrupted) + assert unresolved.state == RESULT_UNKNOWN + assert unresolved.allow_legacy is False + assert len(cancellation_commands) == 1 + + restarted = FakeBase(storage) + restarted.now_utc = interrupted.now_utc + restarted.midnight_utc = interrupted.midnight_utc + activate(restarted) + queried = [] + sent_after_restart = [] + + def query(command_id): + queried.append(command_id) + return cancellation_applied(cancellation_commands[0]) + + def send(command): + sent_after_restart.append(command) + return applied(command) + + install_bindings( + restarted, + storage, + send, + fence=11, + expiry=expiry, + query=query, + ) + decision = before_legacy_plan(restarted) + + assert decision.state == RESULT_APPLIED + assert decision.allow_legacy is False + assert queried == [cancellation_commands[0].command_id] + assert len(sent_after_restart) == 1 + assert "replacement" in sent_after_restart[0].to_wire()["schedule_transition"] + + +def test_corrupt_installed_authority_never_reopens_live_legacy_gate(): + """A corrupt restart snapshot is a write barrier, not evidence of release.""" + storage = FakeStorage() + initial = FakeBase(storage) + activate(initial) + install_bindings(initial, storage, applied) + assert before_legacy_plan(initial).state == RESULT_APPLIED + storage.values[("lattice_schedule_authority", "schedule")]["installed"]["applied_plan_digest"] = "corrupt" + + restarted = FakeBase(storage) + decision = before_legacy_plan(restarted) + + assert decision.state == RESULT_UNKNOWN + assert decision.allow_legacy is False + assert get_schedule_coordinator(restarted).gate.allowed is False + assert "corrupt" in decision.detail + + +def test_marker_load_failure_is_fail_closed_when_durable_state_is_unknowable(): + storage = FakeStorage() + storage.fail_loads = True + restarted = FakeBase(storage) + + decision = before_legacy_plan(restarted) + + assert decision.mode == MODE_OFF + assert decision.allow_legacy is False + assert "could not be loaded" in decision.detail + + +def test_unknown_replacement_preserves_prior_installed_schedule_suppression(): + storage = FakeStorage() + base = FakeBase(storage) + activate(base) + install_bindings(base, storage, applied) + assert before_legacy_plan(base).state == RESULT_APPLIED + + base.charge_limit_best = [9.0] + base._lattice_schedule_coordinator = LatticeScheduleCoordinator(base) + install_bindings(base, storage, unknown, fence=11) + replacement = before_legacy_plan(base) + + assert replacement.allow_legacy is False + assert replacement.state == RESULT_UNKNOWN + durable = storage.values[("lattice_schedule_authority", "schedule")] + assert durable["installed"]["state"] == RESULT_APPLIED + + +def test_rejected_replacement_preserves_prior_installed_schedule_suppression(): + storage = FakeStorage() + base = FakeBase(storage) + activate(base) + install_bindings(base, storage, applied) + assert before_legacy_plan(base).state == RESULT_APPLIED + + base.charge_limit_best = [9.0] + base._lattice_schedule_coordinator = LatticeScheduleCoordinator(base) + install_bindings(base, storage, rejected, fence=11) + replacement = before_legacy_plan(base) + + assert replacement.allow_legacy is False + assert replacement.state == RESULT_NOT_APPLIED + durable = storage.values[("lattice_schedule_authority", "schedule")] + assert durable["installed"]["state"] == RESULT_APPLIED + + +def test_execute_and_balance_stop_at_shared_gate_before_touching_writer_state(): + base = FakeBase() + activate(base) + + result = Execute.execute_plan(base) + balanced = Execute.balance_inverters(base) + + assert result[0] == "Lattice schedule" + assert balanced is False + assert base.status_calls == [(False, False, True)] + + +def test_gateway_plan_republish_and_commands_share_closed_gate(): + base = FakeBase() + activate(base) + gateway = GatewayMQTT.__new__(GatewayMQTT) + gateway.base = base + gateway.log = base.log + gateway._mqtt_connected = True + gateway._mqtt_client = SimpleNamespace(publish=AsyncMock()) + gateway.topic_schedule = "device/schedule" + gateway.topic_command = "device/command" + gateway._last_published_plan = None + gateway._pending_plan = None + gateway._plan_version = 0 + gateway._last_plan_entries = [{"mode": 1}] + gateway._last_plan_timezone = "UTC" + gateway._last_plan_publish_time = 0 + gateway._last_plan_data = None + gateway._command_id = 0 + gateway.build_execution_plan = lambda *_args, **_kwargs: b"legacy-plan" + gateway.build_command = lambda *_args, **_kwargs: "{}" + + asyncio.run(gateway.publish_plan([{"mode": 1}], "UTC")) + asyncio.run(gateway._republish_plan_if_stale()) + asyncio.run(gateway.publish_command("set_charge_rate", power_w=3000)) + + gateway._mqtt_client.publish.assert_not_awaited() + assert gateway._command_id == 0 + + +def test_async_writer_wait_does_not_block_event_loop_and_releases_in_order(): + """A queued Gateway-style writer waits in an executor while the loop runs.""" + base = FakeBase() + entered = asyncio.Event() + release = asyncio.Event() + order = [] + + @lattice_async_writer("test async legacy writer") + async def blocking_writer(subject, name): + order.append(("enter", name)) + if name == "first": + entered.set() + await release.wait() + order.append(("exit", name)) + + async def scenario(): + first = asyncio.create_task(blocking_writer(base, "first")) + await asyncio.wait_for(entered.wait(), timeout=2) + second = asyncio.create_task(blocking_writer(base, "second")) + await asyncio.sleep(0.02) + assert not second.done() + assert order == [("enter", "first")] + release.set() + await asyncio.wait_for(asyncio.gather(first, second), timeout=2) + + asyncio.run(scenario()) + + assert order == [ + ("enter", "first"), + ("exit", "first"), + ("enter", "second"), + ("exit", "second"), + ] + assert get_schedule_coordinator(base).gate.active_writers == 0 + + +def test_complete_plan_adapter_rejects_incomplete_limit_arrays(): + base = FakeBase() + base.charge_limit_best = [] + try: + build_complete_plan(base) + except Exception as exc: + assert "incomplete" in str(exc) + else: + raise AssertionError("incomplete optimizer plan was accepted") From 64cc7143894ba0fb135d3c8d059ea90a8f6ef251 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 28 Jul 2026 09:54:56 +0100 Subject: [PATCH 2/2] fix: exclude lattice coordinator from debug export --- apps/predbat/userinterface.py | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/predbat/userinterface.py b/apps/predbat/userinterface.py index 7e9726cf0..ba822018a 100644 --- a/apps/predbat/userinterface.py +++ b/apps/predbat/userinterface.py @@ -52,6 +52,7 @@ "github_url_cache", "octopus_url_cache", "secrets", + "_lattice_schedule_coordinator", ]