diff --git a/.cspell/custom-dictionary-workspace.txt b/.cspell/custom-dictionary-workspace.txt index fd636a2ec..02a9f559e 100644 --- a/.cspell/custom-dictionary-workspace.txt +++ b/.cspell/custom-dictionary-workspace.txt @@ -241,6 +241,7 @@ foxess foxesscloud freephase fromisoformat +fromlist fronius funcs futurerate @@ -536,6 +537,7 @@ pypi pyproject pytest pythonhashseed +PYTHONPATH pytz Quattro Qwen diff --git a/apps/predbat/control_handoff.py b/apps/predbat/control_handoff.py new file mode 100644 index 000000000..147b4f7a5 --- /dev/null +++ b/apps/predbat/control_handoff.py @@ -0,0 +1,183 @@ +"""Optional process-local write fencing, not a backend handoff acknowledgement. + +Only an explicitly attached base.control_write_gate enables this foundation. +No configuration flag, receipt publisher or controller release is provided yet. +""" + +import asyncio +import functools +import inspect +import math +import threading +import time + + +class ControlWriteBlocked(RuntimeError): + """A closed, expired or wrong-provider epoch cannot admit another mutation.""" + + +class ControlWriteGate: + """Fence transport attempts across component threads and independent event loops. + + Closing admission is permanent. Uncertain remote outcomes remain latched, + even after all local tasks finish. A transport drain is never proof that a + Hub is physically fenced or that a backend transition was committed. + """ + + def __init__(self, provider, epoch, config_version, expires_at, clock=time.monotonic): + """Bind one gate to a controller epoch and a monotonic lease deadline.""" + if not provider or not epoch or isinstance(config_version, bool) or not isinstance(config_version, int): + raise ValueError("Invalid controller binding") + if not math.isfinite(expires_at) or expires_at <= clock(): + raise ValueError("Invalid controller lease") + self.provider = provider + self.epoch = epoch + self.config_version = config_version + self.expires_at = expires_at + self._clock = clock + self._condition = threading.Condition() + self._closed = False + self._transition_id = None + self._next_token = 0 + self._pending = set() + self._uncertain = set() + + def admit(self, provider): + """Atomically admit one transport attempt or reject before transmission.""" + with self._condition: + if self._closed or provider != self.provider or self._clock() >= self.expires_at: + raise ControlWriteBlocked("Controller epoch cannot admit a mutation") + self._next_token += 1 + self._pending.add(self._next_token) + return self._next_token + + def complete(self, token, definitive): + """Finish local work, retaining every remotely ambiguous outcome.""" + with self._condition: + if token not in self._pending: + raise ValueError("Unknown or already completed mutation") + self._pending.remove(token) + if definitive is not True: + self._uncertain.add(token) + self._condition.notify_all() + + def close(self, epoch, config_version, transition_id): + """Permanently close this exact epoch for one transition, idempotently.""" + with self._condition: + if epoch != self.epoch or config_version != self.config_version or not transition_id: + raise ValueError("Wrong controller binding") + if self._transition_id is not None and transition_id != self._transition_id: + raise ValueError("Controller already belongs to another transition") + self._closed = True + self._transition_id = transition_id + self._condition.notify_all() + + def snapshot(self): + """Return local evidence without claiming a durable or physical handoff.""" + with self._condition: + return { + "provider": self.provider, + "epoch": self.epoch, + "config_version": self.config_version, + "transition_id": self._transition_id, + "closed": self._closed, + "expired": self._clock() >= self.expires_at, + "pending": len(self._pending), + "uncertain": len(self._uncertain), + } + + def wait_drained(self, epoch, config_version, transition_id, timeout): + """Wait for local drain only; timeout, expiry and ambiguity return false. + + This synchronous method belongs on a worker thread, never on a component + event loop needed to finish admitted work. Cancelling an async caller + must not cause it to emit an acknowledgement from a finally block. + """ + if not math.isfinite(timeout) or timeout < 0: + raise ValueError("Invalid drain timeout") + deadline = time.monotonic() + timeout + with self._condition: + if not self._closed or self._transition_id != transition_id or epoch != self.epoch or config_version != self.config_version: + raise ValueError("Drain requires the exact closed transition") + while self._pending: + remaining = min(deadline - time.monotonic(), self.expires_at - self._clock()) + if remaining <= 0: + return False + self._condition.wait(remaining) + return not self._uncertain and self._clock() < self.expires_at + + +def control_gate_for(base): + """Use only an explicitly attached gate; ordinary installations are unchanged.""" + gate = getattr(base, "__dict__", {}).get("control_write_gate") + if gate is not None and not isinstance(gate, ControlWriteGate): + raise ControlWriteBlocked("Invalid controller gate") + return gate + + +def control_mutation(provider, endpoint, definitive_result, definitive_error): + """Fence each actual Cloud mutation, including each retry's transport call.""" + + def decorate(function): + """Wrap a transport without changing its public calling convention.""" + signature = inspect.signature(function) + + @functools.wraps(function) + async def guarded(self, *args, **kwargs): + """Track one submitted write until its complete classified outcome.""" + gate = control_gate_for(getattr(self, "base", None)) + if gate is None: + return await function(self, *args, **kwargs) + arguments = signature.bind(self, *args, **kwargs) + arguments.apply_defaults() + if arguments.arguments.get("endpoint") != endpoint or not arguments.arguments.get("post"): + return await function(self, *args, **kwargs) + token = gate.admit(provider) + definitive = False + try: + result = await function(self, *args, **kwargs) + definitive = definitive_result(result) is True + return result + except Exception as error: + definitive = definitive_error(error) is True + raise + finally: + # BaseException (including cancellation) deliberately stays uncertain. + gate.complete(token, definitive) + + return guarded + + return decorate + + +async def gated_mqtt_publish(gate, provider, publish, owner_loop): + """Keep owner-loop work counted even if its awaiting caller is cancelled. + + Broker acceptance is not physical completion. Every submitted MQTT mutation + stays uncertain until a future protocol can prove the physical fence. + """ + token = gate.admit(provider) + + def finished(future): + """Consume completion and retain an unresolved physical outcome.""" + if not future.cancelled(): + future.exception() + gate.complete(token, False) + + coroutine = None + try: + coroutine = publish() + if owner_loop is not None and owner_loop is not asyncio.get_running_loop(): + future = asyncio.run_coroutine_threadsafe(coroutine, owner_loop) + future.add_done_callback(finished) + waiter = asyncio.wrap_future(future) + else: + future = asyncio.ensure_future(coroutine) + future.add_done_callback(finished) + waiter = future + except BaseException: + if coroutine is not None: + coroutine.close() + gate.complete(token, False) + raise + await asyncio.shield(waiter) diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index ee4f99d6c..a7b832cbc 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -21,6 +21,7 @@ import pytz as _pytz from component_base import ComponentBase +from control_handoff import control_gate_for, gated_mqtt_publish try: import gateway_status_pb2 as pb @@ -2617,6 +2618,14 @@ async def _publish_raw(self, topic, payload, retain=False): if not (self._mqtt_client and self._mqtt_connected): return + gate = control_gate_for(getattr(self, "base", None)) + # Battery control only. EV commands are an independent control role. + if gate is not None and topic in {f"{self._topic_base}/command", f"{self._topic_base}/schedule"}: + # Includes retained schedules and owner-loop futures, not just + # ad-hoc commands. No queued plan can bypass a closed epoch later. + await gated_mqtt_publish(gate, "predbat_gateway", lambda: self._mqtt_client.publish(topic, payload, qos=1, retain=retain), self._loop) + return + if self._loop is not None and self._loop is not asyncio.get_running_loop(): future = asyncio.run_coroutine_threadsafe( self._mqtt_client.publish(topic, payload, qos=1, retain=retain), diff --git a/apps/predbat/gecloud.py b/apps/predbat/gecloud.py index 77341d8fd..7ba1d008b 100644 --- a/apps/predbat/gecloud.py +++ b/apps/predbat/gecloud.py @@ -25,6 +25,7 @@ from component_base import ComponentBase from coordinator import inverter_record from mock_base import MockBase as SharedMockBase +from control_handoff import control_mutation """ GE Cloud data download @@ -53,6 +54,22 @@ GE_REGISTER_BATTERY_CUTOFF_LIMIT = 75 + +def ge_write_response_definitive(result): + """Accept only documented single-device success with a valid setting value. + + GivEnergy's Remote Control Codes contract describes this as the device's + resulting value. This is provider completion evidence, never an exclusive + controller handoff acknowledgement. Optimistic pending_writes is not used. + """ + if not isinstance(result, dict) or result.get("success") is not True: + return False + value = result.get("value") + if isinstance(value, bool): + return True + return isinstance(value, (int, float)) and math.isfinite(value) and value >= 0 + + # How long the cached customer account details stay valid for before they are fetched again ACCOUNT_MAX_AGE_MINUTES = 24 * 60 # How long to wait before retrying a failed account fetch @@ -3264,6 +3281,12 @@ async def async_get_inverter_data_retry(self, endpoint, serial="", setting_id="" self.log("GECloud: Warn: Failed to get data from {}".format(endpoint)) return data + @control_mutation( + "givenergy", + GE_API_INVERTER_WRITE_SETTING, + definitive_result=ge_write_response_definitive, + definitive_error=lambda error: isinstance(error, GECloudTerminalError) and error.code in (-3, -4, -7), + ) async def async_get_inverter_data(self, endpoint, serial="", setting_id="", post=False, datain=None, uuid="", meter_ids="", start_time="", end_time="", command="", measurands=""): """ Basic API call to GE Cloud diff --git a/apps/predbat/tests/test_control_handoff.py b/apps/predbat/tests/test_control_handoff.py new file mode 100644 index 000000000..c7649f7ad --- /dev/null +++ b/apps/predbat/tests/test_control_handoff.py @@ -0,0 +1,400 @@ +"""Synthetic races at the actual GivEnergy and Gateway mutation sinks.""" + +import asyncio +import threading +import time +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +from control_handoff import ControlWriteBlocked, ControlWriteGate +from gecloud import GECloudDirect, GE_API_INVERTER_READ_SETTING, GE_API_INVERTER_WRITE_SETTING +from gateway import GatewayMQTT + + +def make_gate(provider="givenergy", clock=time.monotonic): + """Make one synthetic controller with no external backend or credentials.""" + return ControlWriteGate(provider, "epoch-1", 7, clock() + 60, clock=clock) + + +def close_gate(gate): + """Close exactly the synthetic source epoch.""" + gate.close("epoch-1", 7, "transition-1") + + +def drained(gate, timeout=0): + """Read local transport-drain evidence for the synthetic binding.""" + return gate.wait_drained("epoch-1", 7, "transition-1", timeout) + + +def make_cloud(gate=None): + """Construct the actual transport class without starting component threads.""" + cloud = GECloudDirect.__new__(GECloudDirect) + cloud.base = SimpleNamespace() + if gate is not None: + cloud.base.control_write_gate = gate + cloud.api_key = "synthetic-key" + cloud.requests_total = 0 + cloud.failures_total = 0 + cloud.api_auth_failed = False + cloud.pending_writes = {"A": []} + cloud.log = lambda *_args, **_kwargs: None + cloud.update_success_timestamp = lambda: None + return cloud + + +class FakeResponse: + """A controlled provider response, reached through the real aiohttp call site.""" + + status = 200 + + def __init__(self, outcome): + """Save an awaitable body supplier.""" + self.outcome = outcome + + async def __aenter__(self): + """Enter the synthetic response context.""" + return self + + async def __aexit__(self, *_args): + """Leave the response context without consuming exceptions.""" + return False + + async def json(self): + """Return or fail the body only when the test permits it.""" + return await self.outcome() + + +class FakeSession: + """Intercept real transport requests without contacting a customer or provider.""" + + def __init__(self, outcome): + """Collect every submitted URL and body.""" + self.outcome = outcome + self.requests = [] + + async def __aenter__(self): + """Enter a synthetic aiohttp session.""" + return self + + async def __aexit__(self, *_args): + """Leave the session.""" + return False + + def post(self, url, **kwargs): + """Record an actual POST admission.""" + self.requests.append((url, kwargs)) + return FakeResponse(self.outcome) + + +def make_gateway(gate, client, loop=None): + """Create the actual MQTT component without opening a broker connection.""" + gateway = GatewayMQTT.__new__(GatewayMQTT) + gateway.base = SimpleNamespace(control_write_gate=gate) if gate is not None else SimpleNamespace() + gateway.gateway_display_only = False + gateway._topic_base = "predbat/devices/pbgw_synthetic" + gateway.topic_command = gateway._topic_base + "/command" + gateway.topic_schedule = gateway._topic_base + "/schedule" + gateway._mqtt_client = client + gateway._mqtt_connected = True + gateway._loop = loop + gateway.log = lambda *_args, **_kwargs: None + return gateway + + +class TestControlWriteGate(unittest.TestCase): + """Thread-safe epoch/revision binding, timeouts and uncertainty semantics.""" + + def test_binding_timeout_expiry_and_permanent_close(self): + """No stale binding, wait timeout or expired lease can authorise drain.""" + clock = [100.0] + gate = make_gate(clock=lambda: clock[0]) + token = gate.admit("givenergy") + with self.assertRaises(ControlWriteBlocked): + gate.admit("fox_ess") + with self.assertRaises(ValueError): + gate.close("old-epoch", 7, "transition-1") + close_gate(gate) + close_gate(gate) + with self.assertRaises(ValueError): + gate.close("epoch-1", 7, "another-transition") + self.assertFalse(drained(gate)) + gate.complete(token, True) + self.assertTrue(drained(gate)) + with self.assertRaises(ControlWriteBlocked): + gate.admit("givenergy") + clock[0] += 61 + self.assertFalse(drained(gate)) + with self.assertRaises(ValueError): + gate.complete(token, True) + + def test_concurrent_admission_and_closure(self): + """Race eight real threads against permanent source fencing.""" + gate = make_gate() + barrier = threading.Barrier(9) + outcomes = [] + + def worker(): + """Attempt exactly one mutation across the close boundary.""" + barrier.wait() + try: + token = gate.admit("givenergy") + except ControlWriteBlocked: + outcomes.append("blocked") + else: + gate.complete(token, True) + outcomes.append("finished") + + threads = [threading.Thread(target=worker) for _ in range(8)] + for thread in threads: + thread.start() + barrier.wait() + close_gate(gate) + for thread in threads: + thread.join(2) + self.assertFalse(thread.is_alive()) + self.assertEqual(len(outcomes), 8) + self.assertTrue(drained(gate)) + with self.assertRaises(ControlWriteBlocked): + gate.admit("givenergy") + + +class TestWiredCloudGate(unittest.IsolatedAsyncioTestCase): + """Exercise the decorated production transport and its production retry caller.""" + + async def test_pending_write_closes_new_admission_but_allows_post_reads(self): + """Documented device success completes a local drain; reads still work.""" + gate = make_gate() + cloud = make_cloud(gate) + started, release = asyncio.Event(), asyncio.Event() + + async def outcome(): + """Hold one provider response across gate closure.""" + started.set() + await release.wait() + return {"data": {"value": 1, "success": True}} + + session = FakeSession(outcome) + with patch("gecloud.aiohttp.ClientSession", return_value=session): + write = asyncio.create_task(cloud.async_get_inverter_data(GE_API_INVERTER_WRITE_SETTING, "A", 1, post=True, datain={"value": "1"})) + await started.wait() + close_gate(gate) + self.assertEqual(gate.snapshot()["pending"], 1) + self.assertFalse(drained(gate)) + with self.assertRaises(ControlWriteBlocked): + await cloud.async_get_inverter_data(GE_API_INVERTER_WRITE_SETTING, "A", 1, post=True, datain={"value": "2"}) + self.assertEqual(len(session.requests), 1) + release.set() + self.assertEqual(await write, {"value": 1, "success": True}) + self.assertTrue(drained(gate)) + self.assertEqual(gate.snapshot()["uncertain"], 0) + await cloud.async_get_inverter_data(GE_API_INVERTER_READ_SETTING, "A", 1, post=True) + self.assertEqual(len(session.requests), 2) + + async def test_strict_vendor_success_and_truthy_post(self): + """Malformed, missing and ambiguous results never become completion evidence.""" + cases = [ + ({"value": True, "success": True}, True), + ({"value": 12.5, "success": True}, True), + ({"value": 1}, False), + ({"value": 1, "success": 1}, False), + ({"value": "1", "success": True}, False), + ({"success": True}, False), + ({"value": float("nan"), "success": True}, False), + ({"value": float("inf"), "success": True}, False), + ({"value": -1, "success": False}, False), + ({"value": -5, "success": False}, False), + ({"value": -6, "success": False}, False), + ] + for body, expected in cases: + with self.subTest(body=body): + gate = make_gate() + cloud = make_cloud(gate) + + async def outcome(): + """Supply one unwrapped vendor response shape.""" + return {"data": body} + + session = FakeSession(outcome) + with patch("gecloud.aiohttp.ClientSession", return_value=session): + await cloud.async_get_inverter_data(GE_API_INVERTER_WRITE_SETTING, "A", 1, 1) + close_gate(gate) + self.assertEqual(drained(gate), expected) + with self.assertRaises(ControlWriteBlocked): + await cloud.async_get_inverter_data(GE_API_INVERTER_WRITE_SETTING, "A", 1, 1) + self.assertEqual(len(session.requests), 1) + + async def test_retry_never_submits_after_closure(self): + """Vendor timeout uncertainty remains even though subsequent retries are fenced.""" + gate = make_gate() + cloud = make_cloud(gate) + + async def outcome(): + """Represent a potentially successful vendor timeout code.""" + return {"data": {"value": -1}} + + async def stop_retry(_delay): + """Fence while the actual retry loop is backing off.""" + close_gate(gate) + + session = FakeSession(outcome) + with patch("gecloud.aiohttp.ClientSession", return_value=session), patch("gecloud.asyncio.sleep", side_effect=stop_retry): + with self.assertRaises(ControlWriteBlocked): + await cloud.async_write_inverter_setting("A", 1, 1) + self.assertEqual(len(session.requests), 1) + self.assertEqual(gate.snapshot()["uncertain"], 1) + self.assertFalse(drained(gate)) + + async def test_timeout_late_application_never_claims_drained(self): + """A server operation outlives its timed-out client; local idle is insufficient.""" + gate = make_gate() + cloud = make_cloud(gate) + apply_later = asyncio.Event() + applied = [] + + async def server(): + """Model remote work that cannot be cancelled by the client timeout.""" + await apply_later.wait() + applied.append(1) + + async def outcome(): + """The transport times out before the provider operation finishes.""" + raise asyncio.TimeoutError() + + server_task = asyncio.create_task(server()) + with patch("gecloud.aiohttp.ClientSession", return_value=FakeSession(outcome)): + self.assertIsNone(await cloud.async_get_inverter_data(GE_API_INVERTER_WRITE_SETTING, "A", 1, post=True, datain={"value": "1"})) + close_gate(gate) + self.assertEqual(gate.snapshot()["pending"], 0) + self.assertEqual(gate.snapshot()["uncertain"], 1) + self.assertFalse(drained(gate)) + apply_later.set() + await server_task + self.assertEqual(applied, [1]) + self.assertFalse(drained(gate)) + + async def test_cancelled_transport_is_ambiguous(self): + """Cancellation never becomes a positive source-drained receipt.""" + gate = make_gate() + cloud = make_cloud(gate) + started = asyncio.Event() + + async def outcome(): + """Hold the actual write response until its client is cancelled.""" + started.set() + await asyncio.Event().wait() + + with patch("gecloud.aiohttp.ClientSession", return_value=FakeSession(outcome)): + task = asyncio.create_task(cloud.async_get_inverter_data(GE_API_INVERTER_WRITE_SETTING, "A", 1, post=True, datain={"value": "1"})) + await started.wait() + close_gate(gate) + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + self.assertFalse(drained(gate)) + self.assertEqual(gate.snapshot()["uncertain"], 1) + + async def test_no_gate_preserves_normal_transport(self): + """Normal installations require no transition configuration.""" + cloud = make_cloud() + + async def outcome(): + """Return the normal successful write response.""" + return {"data": {"value": 1}} + + session = FakeSession(outcome) + with patch("gecloud.aiohttp.ClientSession", return_value=session): + self.assertEqual(await cloud.async_get_inverter_data(GE_API_INVERTER_WRITE_SETTING, "A", 1, post=True, datain={"value": "1"}), {"value": 1}) + self.assertEqual(len(session.requests), 1) + + +class TestWiredGatewayGate(unittest.IsolatedAsyncioTestCase): + """MQTT owner-loop futures, retained schedules and optional display traffic.""" + + async def test_owner_loop_future_survives_waiter_cancellation_and_blocks_drain(self): + """Keep a submitted retained schedule counted until the owning loop finishes.""" + gate = make_gate("predbat_gateway") + started, release = threading.Event(), threading.Event() + calls = [] + loop = asyncio.new_event_loop() + thread = threading.Thread(target=loop.run_forever) + thread.start() + + async def publish(topic, payload, **kwargs): + """Model a pending broker publish on its actual owning thread.""" + calls.append((topic, payload, kwargs)) + started.set() + while not release.is_set(): + await asyncio.sleep(0.001) + + gateway = make_gateway(gate, SimpleNamespace(publish=publish), loop) + try: + task = asyncio.create_task(gateway._publish_raw(gateway.topic_schedule, b"schedule", retain=True)) + self.assertTrue(await asyncio.to_thread(started.wait, 2)) + close_gate(gate) + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + self.assertEqual(gate.snapshot()["pending"], 1) + self.assertFalse(drained(gate)) + with self.assertRaises(ControlWriteBlocked): + await gateway._publish_raw(gateway.topic_command, b"command") + release.set() + for _ in range(1000): + if gate.snapshot()["pending"] == 0: + break + await asyncio.sleep(0.001) + self.assertEqual(gate.snapshot()["pending"], 0) + self.assertEqual(gate.snapshot()["uncertain"], 1) + self.assertFalse(drained(gate)) + self.assertEqual(len(calls), 1) + self.assertTrue(calls[0][2]["retain"]) + await gateway._publish_raw(gateway._topic_base + "/predbat_data", b"display") + self.assertEqual(len(calls), 2) + await gateway._publish_raw(gateway._topic_base + "/query", b"read-only-query") + self.assertEqual(len(calls), 3) + await gateway._publish_raw(gateway._topic_base + "/ev/command", b"independent-charger") + self.assertEqual(len(calls), 4) + finally: + release.set() + loop.call_soon_threadsafe(loop.stop) + await asyncio.to_thread(thread.join, 2) + loop.close() + + async def test_queued_retained_plan_cannot_escape_closed_epoch(self): + """The actual reconnect and periodic schedule paths reach the fenced sink.""" + gate = make_gate("predbat_gateway") + calls = [] + + async def publish(*args, **kwargs): + """Collect only publishes that reached the transport.""" + calls.append((args, kwargs)) + + gateway = make_gateway(gate, SimpleNamespace(publish=publish)) + gateway._mqtt_connected = False + gateway._last_published_plan = None + gateway._pending_plan = None + await gateway.publish_plan([], "UTC") + self.assertEqual(gateway._pending_plan, ([], "UTC")) + close_gate(gate) + gateway._mqtt_connected = True + gateway._plan_version = 1 + gateway.build_execution_plan = lambda *_args, **_kwargs: b"schedule" + gateway._debug_dump = lambda *_args, **_kwargs: None + with self.assertRaises(ControlWriteBlocked): + await gateway.publish_plan(*gateway._pending_plan) + gateway._last_plan_publish_time = 0 + with self.assertRaises(ControlWriteBlocked): + await gateway._republish_plan_if_stale() + self.assertEqual(calls, []) + + +def run_control_handoff_tests(_predbat=None): + """Run through Predbat's normal unit-test registry as well as unittest.""" + suite = unittest.defaultTestLoader.loadTestsFromModule(__import__(__name__, fromlist=["_"])) + result = unittest.TextTestRunner(verbosity=2).run(suite) + return not result.wasSuccessful() + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/predbat/unit_test.py b/apps/predbat/unit_test.py index 2f2bf0268..aad78b77b 100644 --- a/apps/predbat/unit_test.py +++ b/apps/predbat/unit_test.py @@ -103,6 +103,7 @@ from tests.test_teslemetry import test_teslemetry from tests.test_compare import test_compare from tests.test_gateway import run_gateway_tests +from tests.test_control_handoff import run_control_handoff_tests from tests.test_axle import test_axle from tests.test_db_manager import test_db_manager from tests.test_hahistory import run_hahistory_tests @@ -753,6 +754,7 @@ def main(): ("load_today_comparison", test_load_today_comparison, "load_today_comparison None-guard regression test", False), ("compare", test_compare, "Compare tariff engine tests (hardware overrides, bleed isolation)", False), ("gateway", run_gateway_tests, "GatewayMQTT component tests (protobuf, plan serialization, commands, telemetry)", False), + ("control_handoff", run_control_handoff_tests, "Epoch-bound transport fencing and drain races (Cloud and Gateway)", False), ("optimise_levels", run_optimise_levels_tests, "Optimise levels tests", False), ("trim_export", run_trim_export_tests, "Export trim ordering (buffer from cheapest slot) tests", False), ("plan_tiebreak", run_plan_tiebreak_tests, "Plan fragmentation near-tie tie-break tests", False), diff --git a/docs/control-write-drain-gate.md b/docs/control-write-drain-gate.md new file mode 100644 index 000000000..2cae3fd3c --- /dev/null +++ b/docs/control-write-drain-gate.md @@ -0,0 +1,85 @@ +# Battery control write fencing foundation + +This patch does not enable a control selector, automatic backup or a production +handoff. There is no backend acknowledgement publisher, runtime configuration +switch or controller release. Existing installations have no gate and retain +their current behaviour. + +An explicitly attached `base.control_write_gate` binds battery control to one +provider, epoch, configuration revision and monotonic lease. Admission and +closure share a thread lock. Closure is permanent and transition-bound. Every +actual GivEnergy inverter setting POST, including retries, passes admission. +Gateway battery commands and executable schedules pass admission immediately +before MQTT submission, including reconnect and periodic retained-plan paths. +Expired epochs reject new attempts. Neither a wait timeout nor cancellation +proves drain. Previously ambiguous attempts remain latched for that epoch. + +## Provider completion evidence + +The [GivEnergy API Remote Control Codes contract](https://givenergy.cloud/docs/api/v1#remote-control-codes) +describes a single-device response with a success boolean and the resulting +device value. We infer completion of that individual setting operation only +when `success is True` and the value is a finite nonnegative number or boolean. +Strings, missing fields, malformed success flags, negative values and missing +responses remain uncertain. This deliberately supports a conservative subset +of setting value types. Terminal exceptions for documented rejection codes +-3, -4 and -7 are definitive non-application. Retryable errors remain uncertain, +including -2 despite its documented non-application, because it is currently +handled in the shared retry-body path. Successful later attempts never clear +an earlier uncertain attempt. The optimistic `pending_writes` cache is not +physical readback evidence and is not used by the gate. + +MQTT publication completion proves broker acceptance only. Every submitted +battery command or schedule therefore remains uncertain even after its owner +loop future completes. A caller cancellation leaves that future tracked until +it actually completes. A retained schedule already accepted by the broker or +Hub is not revoked by closing this gate and may execute later. The gate cannot +currently certify a physical Hub handoff. + +The supported domain is battery inverter control. Gateway `ev/command` and +GivEnergy charger endpoints are independent roles and remain available, as do +Hub display data and query topics. Other inverter providers need their own +transport adapters and completion contracts before supporting handoff. + +## Remaining complete handoff contract + +A canary-gated coordinator must durably bind a transition to source epoch, +configuration revision, exact inverter identity and target provider. The source +must close admission before returning evidence. Replayed, expired and mismatched +receipts must be rejected. A fetched configuration revision is not evidence +that the old writer stopped. Process restarts must reconstruct closed epochs +rather than silently reverting to a controller with no gate. + +The Hub protocol must acknowledge an epoch/revision-bound physical fence after +old commands have completed and old retained schedules have been invalidated. +Read-only probe leases must be recognised by standby reconciliation and expire +back to parked state. Target initialisation and all write paths must remain +blocked until source release and fresh target readiness are verified. Only then +may the backend commit the provider change and release the target writer. +No such physical fence or durable receipt exists in this patch, so activation +must remain unavailable. + +`download.get_github_directory_listing` enumerates all files under +`apps/predbat`, so `control_handoff.py` participates in the existing dynamic +module download. Bundled images must include that new file with both importing +modules. No separate hardcoded module manifest was found in this repository. + +## Synthetic verification + +Python 3.14.8, invoked through `/private/tmp/predbat-hub-tools-venv/bin/python` +(which resolves to the Homebrew interpreter), runs the actual decorated +`GECloudDirect` and actual `GatewayMQTT` sinks with synthetic HTTP and MQTT +transports. Tests race eight threads, close while a POST is outstanding, reject +truthy positional POST retries after closure, validate strict vendor responses, +retain timeout/late-server-application uncertainty, cancel an awaiting Gateway +caller while its owner-loop future is pending, and fence retained-plan reconnect +and periodic republish. Independent EV, display and query traffic remains usable. +The existing Gateway and GivEnergy suites also run unchanged. + +From `coverage`, run each registered suite with: + +```sh +env PYTHONPATH=../apps/predbat /private/tmp/predbat-hub-tools-venv/bin/python ../apps/predbat/unit_test.py --test control_handoff +env PYTHONPATH=../apps/predbat /private/tmp/predbat-hub-tools-venv/bin/python ../apps/predbat/unit_test.py --test gateway +env PYTHONPATH=../apps/predbat /private/tmp/predbat-hub-tools-venv/bin/python ../apps/predbat/unit_test.py --test ge_cloud +```