From ab86dc3b0bab4de165b8864c6744741b92490a6d Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 28 Jul 2026 07:31:42 +0100 Subject: [PATCH] feat(lattice): add fragment adapter registry --- apps/predbat/lattice_fragment_adapters.py | 535 ++++++++++++++++++ .../tests/test_lattice_fragment_adapters.py | 520 +++++++++++++++++ 2 files changed, 1055 insertions(+) create mode 100644 apps/predbat/lattice_fragment_adapters.py create mode 100644 apps/predbat/tests/test_lattice_fragment_adapters.py diff --git a/apps/predbat/lattice_fragment_adapters.py b/apps/predbat/lattice_fragment_adapters.py new file mode 100644 index 000000000..e8c4623d1 --- /dev/null +++ b/apps/predbat/lattice_fragment_adapters.py @@ -0,0 +1,535 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - Lattice fragment adapter registry +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Generic durable integration adapters for compiled Lattice fragments. + +This module is deliberately additive and default-off. Nothing discovers or +registers production integrations unless a caller explicitly enables a +``FragmentAdapterRegistry`` and asks it to create a compiler. + +An integration owns one ``DurableFragmentAdapter`` and its durable state store. +The adapter atomically binds every generation to one semantic fingerprint and +one immutable ``ProviderSnapshot``. The registry only discovers the common +``lattice_fragment_adapter()`` surface; it has no provider or brand allow-list. +""" + +# cspell:ignore autoconfig idempotently unsubscribers + +import hashlib +import threading +from dataclasses import dataclass +from types import MappingProxyType +from typing import Optional, Protocol + +from lattice_autoconfig import ( + ProviderHealth, + ProviderSnapshot, + _fingerprint_snapshot, + _plain, +) +from lattice_compiled_publication import CompiledLatticeCompiler + + +class FragmentAdapterError(RuntimeError): + """Base error for fail-closed fragment adapter operations.""" + + +class FragmentAdapterReadError(FragmentAdapterError): + """A durable fragment could not be read or validated.""" + + +class FragmentAdapterConflict(FragmentAdapterError): + """An atomic fragment publication lost to a different durable state.""" + + +class FragmentAdapterRemoved(FragmentAdapterReadError): + """A registered integration has published a durable removal tombstone.""" + + +def _validate_provider_id(provider_id): + """Normalize one provider-owned stable identifier.""" + if not isinstance(provider_id, str) or not provider_id.strip(): + raise ValueError("provider_id must be a non-empty string") + return provider_id.strip() + + +def _validate_generation(generation): + """Validate one monotonically increasing integration generation.""" + if not isinstance(generation, int) or isinstance(generation, bool) or generation < 0: + raise ValueError("generation must be a non-negative integer") + + +def _validate_reason(reason): + """Normalize one auditable invalidation reason.""" + if not isinstance(reason, str) or not reason.strip(): + raise ValueError("reason must be a non-empty string") + return reason.strip() + + +def _semantic_fingerprint(snapshot, removed=False): + """Return the compiler's generation-bound semantic safety fingerprint.""" + fingerprint = _fingerprint_snapshot(snapshot) + if removed: + return hashlib.sha256("removed:{}".format(fingerprint).encode("utf-8")).hexdigest() + return fingerprint + + +@dataclass(frozen=True) +class FragmentAdapterState: + """One integration-owned durable fragment cursor and immutable value. + + ``semantic_fingerprint`` intentionally uses the compiler's exact + generation-bound fingerprint. The pair therefore detects both generation + regression and reuse of one generation for different safety-relevant + content. + """ + + provider_id: str + generation: int + semantic_fingerprint: str + snapshot: ProviderSnapshot + removed: bool = False + + def __post_init__(self): + """Validate that the durable cursor exactly binds its snapshot.""" + provider_id = _validate_provider_id(self.provider_id) + _validate_generation(self.generation) + if not isinstance(self.snapshot, ProviderSnapshot): + raise ValueError("snapshot must be ProviderSnapshot") + if self.snapshot.provider_id != provider_id: + raise ValueError("snapshot provider_id does not match durable state") + if self.snapshot.generation != self.generation: + raise ValueError("snapshot generation does not match durable state") + if not isinstance(self.removed, bool): + raise ValueError("removed must be a boolean") + expected = _semantic_fingerprint(self.snapshot, self.removed) + if self.semantic_fingerprint != expected: + raise ValueError("semantic_fingerprint does not match the immutable snapshot") + object.__setattr__(self, "provider_id", provider_id) + + +class FragmentAdapterStateStore: + """Required atomic durable-store protocol owned by one integration.""" + + def load(self): + """Return the current ``FragmentAdapterState`` or ``None``.""" + raise NotImplementedError + + def compare_and_store(self, expected, replacement): + """Atomically store replacement only when current state equals expected.""" + raise NotImplementedError + + +class FragmentPublisher(Protocol): + """Structural integration protocol discovered without provider knowledge.""" + + provider_id: str + + def read_state(self) -> FragmentAdapterState: + """Fresh-read the integration-owned durable state.""" + ... + + def read_snapshot(self) -> ProviderSnapshot: + """Fresh-read the current immutable provider snapshot.""" + ... + + def subscribe_invalidation(self, listener): + """Attach the registry's invalidation sink and return an unsubscribe.""" + ... + + +class FragmentPublishingComponent(Protocol): + """Structural component discovery surface used by the generic registry.""" + + def lattice_fragment_adapter(self) -> Optional[FragmentPublisher]: + """Return this component's fragment publisher, if it has one.""" + ... + + +class InMemoryFragmentAdapterStateStore(FragmentAdapterStateStore): + """Thread-safe reference state store for tests; not production durability.""" + + def __init__(self, state=None): + """Create a store optionally seeded with one validated state.""" + if state is not None and not isinstance(state, FragmentAdapterState): + raise ValueError("initial state must be FragmentAdapterState or None") + self._lock = threading.RLock() + self._state = state + self._writes = 0 + + @property + def writes(self): + """Return the number of successful atomic writes.""" + with self._lock: + return self._writes + + def load(self): + """Return the immutable current state.""" + with self._lock: + return self._state + + def compare_and_store(self, expected, replacement): + """Atomically compare the complete cursor and install replacement.""" + if replacement is not None and not isinstance( + replacement, + FragmentAdapterState, + ): + raise ValueError("replacement must be FragmentAdapterState or None") + with self._lock: + if self._state != expected: + return False + self._state = replacement + self._writes += 1 + return True + + +class DurableFragmentAdapter: + """Generic publisher over one integration-owned atomic state store.""" + + def __init__(self, provider_id, state_store): + """Restore one provider cursor without performing discovery or writes.""" + self.provider_id = _validate_provider_id(provider_id) + if not callable(getattr(state_store, "load", None)) or not callable(getattr(state_store, "compare_and_store", None)): + raise ValueError("state_store must provide load and compare_and_store") + self._state_store = state_store + self._lock = threading.RLock() + self._listeners = [] + self._validate_loaded_state(self._load()) + + def _load(self): + """Load durable state and convert store faults into adapter faults.""" + try: + return self._state_store.load() + except Exception as exc: + raise FragmentAdapterReadError( + "durable fragment load failed: {}: {}".format( + type(exc).__name__, + exc, + ) + ) from exc + + def _validate_loaded_state(self, state): + """Validate one store result and its provider ownership.""" + if state is None: + return None + if not isinstance(state, FragmentAdapterState): + raise FragmentAdapterReadError("state_store.load must return FragmentAdapterState or None") + try: + state.__post_init__() + except ValueError as exc: + raise FragmentAdapterReadError(str(exc)) from exc + if state.provider_id != self.provider_id: + raise FragmentAdapterReadError("durable state belongs to provider {}".format(state.provider_id)) + return state + + def read_state(self): + """Fresh-read the complete immutable durable fragment state.""" + with self._lock: + state = self._validate_loaded_state(self._load()) + if state is None: + raise FragmentAdapterReadError("provider {} has no durable fragment".format(self.provider_id)) + return state + + def read_snapshot(self): + """Fresh-read one immutable snapshot or fail closed on removal.""" + state = self.read_state() + if state.removed: + raise FragmentAdapterRemoved( + "provider {} was removed at generation {}".format( + self.provider_id, + state.generation, + ) + ) + return state.snapshot + + def subscribe_invalidation(self, listener): + """Subscribe the compiler-facing invalidation sink.""" + if not callable(listener): + raise ValueError("invalidation listener must be callable") + with self._lock: + if listener in self._listeners: + raise ValueError("invalidation listener is already subscribed") + self._listeners.append(listener) + closed = [False] + + def unsubscribe(): + """Detach this exact listener idempotently.""" + with self._lock: + if closed[0]: + return + closed[0] = True + if listener in self._listeners: + self._listeners.remove(listener) + + return unsubscribe + + def publish( + self, + snapshot, + reason, + feedback_token=None, + removed=False, + ): + """Atomically publish a newer fragment and notify every subscriber. + + Invalidation is announced before the durable CAS. Consequently a + failed or conflicting store write leaves the compiler pending and its + fresh reader behind the requested generation, which fails closed. + A subscriber may return ``False`` only to suppress publication-origin + feedback; that token then causes neither persistence nor recompilation. + """ + if not isinstance(snapshot, ProviderSnapshot): + raise ValueError("snapshot must be ProviderSnapshot") + if snapshot.provider_id != self.provider_id: + raise ValueError("snapshot provider_id does not match adapter") + if not isinstance(removed, bool): + raise ValueError("removed must be a boolean") + reason = _validate_reason(reason) + candidate = FragmentAdapterState( + self.provider_id, + snapshot.generation, + _semantic_fingerprint(snapshot, removed), + snapshot, + removed, + ) + + with self._lock: + current = self._validate_loaded_state(self._load()) + if current is not None: + if candidate.generation < current.generation: + raise ValueError( + "fragment generation {} regressed from {}".format( + candidate.generation, + current.generation, + ) + ) + if candidate.generation == current.generation: + if candidate.semantic_fingerprint != (current.semantic_fingerprint): + raise ValueError("fragment generation {} was reused with different " "content".format(candidate.generation)) + return False + listeners = tuple(self._listeners) + + for listener in listeners: + accepted = listener( + self.provider_id, + candidate.generation, + reason, + feedback_token, + ) + if accepted is False: + return False + + try: + committed = self._state_store.compare_and_store( + current, + candidate, + ) + except Exception as exc: + raise FragmentAdapterConflict( + "durable fragment publication failed: {}: {}".format( + type(exc).__name__, + exc, + ) + ) from exc + if committed is not True: + winner = self._validate_loaded_state(self._load()) + if winner == candidate: + return False + raise FragmentAdapterConflict("durable fragment cursor changed before atomic publication") + return True + + def remove(self, generation, reason, feedback_token=None): + """Publish a durable removal tombstone and invalidate the compiler.""" + _validate_generation(generation) + current = self.read_state() + snapshot = current.snapshot + tombstone = ProviderSnapshot( + self.provider_id, + generation, + ProviderHealth.OFFLINE, + _plain(snapshot.topology_fragment), + snapshot.aliases, + snapshot.identity_aliases, + snapshot.role_assignments, + snapshot.config_projections, + ) + return self.publish( + tombstone, + reason, + feedback_token=feedback_token, + removed=True, + ) + + +class FragmentAdapterRegistry: + """Default-off brand-neutral discovery and frozen compiler registry.""" + + DISCOVERY_METHOD = "lattice_fragment_adapter" + + def __init__(self, enabled=False): + """Create an empty registry; disabled is the safe default.""" + if not isinstance(enabled, bool): + raise ValueError("enabled must be a boolean") + self._enabled = enabled + self._lock = threading.RLock() + self._adapters = {} + self._compiler = None + self._unsubscribers = () + self._sealed = False + + @property + def enabled(self): + """Return whether explicit fragment discovery is enabled.""" + return self._enabled + + @property + def provider_ids(self): + """Return registered provider identities in deterministic order.""" + with self._lock: + return tuple(sorted(self._adapters)) + + @property + def readers(self): + """Return an immutable provider-reader mapping for inspection/tests.""" + with self._lock: + return MappingProxyType({provider_id: adapter.read_snapshot for provider_id, adapter in self._adapters.items()}) + + def _validate_adapter(self, adapter): + """Validate the common adapter surface and its durable current state.""" + provider_id = _validate_provider_id(getattr(adapter, "provider_id", None)) + for method_name in ( + "read_state", + "read_snapshot", + "subscribe_invalidation", + ): + if not callable(getattr(adapter, method_name, None)): + raise ValueError("fragment adapter must provide {}".format(method_name)) + state = adapter.read_state() + if not isinstance(state, FragmentAdapterState): + raise ValueError("adapter read_state must return FragmentAdapterState") + if state.provider_id != provider_id: + raise ValueError("adapter state belongs to provider {}".format(state.provider_id)) + return provider_id + + def discover(self, components): + """Discover every component implementing the common publisher surface.""" + if not self._enabled: + return () + candidates = [] + for component in tuple(components): + factory = getattr(component, self.DISCOVERY_METHOD, None) + if factory is None: + continue + if not callable(factory): + raise ValueError("{} must be callable".format(self.DISCOVERY_METHOD)) + adapter = factory() + if adapter is not None: + candidates.append(adapter) + + validated = [] + seen = set() + for adapter in candidates: + provider_id = self._validate_adapter(adapter) + if provider_id in seen: + raise ValueError("provider {} was discovered more than once".format(provider_id)) + seen.add(provider_id) + validated.append((provider_id, adapter)) + + with self._lock: + if self._sealed: + raise RuntimeError("fragment registry membership is sealed") + duplicate = sorted(provider_id for provider_id, _adapter in validated if provider_id in self._adapters) + if duplicate: + raise ValueError("provider {} is already registered".format(duplicate[0])) + for provider_id, adapter in validated: + self._adapters[provider_id] = adapter + return tuple(provider_id for provider_id, _adapter in validated) + + def register(self, adapter): + """Register one explicitly supplied generic adapter before sealing.""" + if not self._enabled: + return False + provider_id = self._validate_adapter(adapter) + with self._lock: + if self._sealed: + raise RuntimeError("fragment registry membership is sealed") + if provider_id in self._adapters: + raise ValueError("provider {} is already registered".format(provider_id)) + self._adapters[provider_id] = adapter + return True + + def unregister(self, provider_id): + """Remove only pre-bind registration; runtime removal is a tombstone.""" + if not self._enabled: + return False + provider_id = _validate_provider_id(provider_id) + with self._lock: + if self._sealed: + raise RuntimeError("runtime unregister is unsafe; publish a durable removal " "tombstone") + if provider_id not in self._adapters: + raise KeyError("unknown provider {}".format(provider_id)) + del self._adapters[provider_id] + return True + + def create_compiler(self, state_store, override_reader=None): + """Freeze membership and create the sole compiled-Lattice coordinator.""" + if not self._enabled: + raise RuntimeError("fragment adapter registry is disabled") + with self._lock: + if self._sealed: + raise RuntimeError("fragment registry membership is already sealed") + if not self._adapters: + raise RuntimeError("cannot create a compiler without fragment adapters") + for adapter in self._adapters.values(): + self._validate_adapter(adapter) + readers = {provider_id: adapter.read_snapshot for provider_id, adapter in self._adapters.items()} + compiler = CompiledLatticeCompiler( + readers, + state_store=state_store, + override_reader=override_reader, + ) + + unsubscribers = [] + try: + for provider_id, adapter in sorted(self._adapters.items()): + + def invalidate( + source_id, + generation, + reason, + feedback_token, + expected_id=provider_id, + ): + """Forward this adapter's invalidation to the compiler.""" + if source_id != expected_id: + raise FragmentAdapterError( + "adapter {} emitted invalidation for {}".format( + expected_id, + source_id, + ) + ) + accepted = compiler.invalidate( + source_id, + generation, + reason, + feedback_token, + ) + if feedback_token is not None and accepted is False: + return False + return True + + unsubscribe = adapter.subscribe_invalidation(invalidate) + if not callable(unsubscribe): + raise ValueError("subscribe_invalidation must return an " "unsubscribe callable") + unsubscribers.append(unsubscribe) + except Exception: + for unsubscribe in reversed(unsubscribers): + unsubscribe() + raise + + self._compiler = compiler + self._unsubscribers = tuple(unsubscribers) + self._sealed = True + return compiler diff --git a/apps/predbat/tests/test_lattice_fragment_adapters.py b/apps/predbat/tests/test_lattice_fragment_adapters.py new file mode 100644 index 000000000..b0de1b4f0 --- /dev/null +++ b/apps/predbat/tests/test_lattice_fragment_adapters.py @@ -0,0 +1,520 @@ +"""Tests for generic durable Lattice fragment adapter discovery.""" + +# cspell:ignore autoconfig + +import os +import sys +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from lattice_autoconfig import ( # noqa: E402 + CompileStatus, + ProviderHealth, +) +from lattice_compiled_publication import ( # noqa: E402 + InMemoryCompiledLatticeStateStore, +) +from lattice_fragment_adapters import ( # noqa: E402 + DurableFragmentAdapter, + FragmentAdapterConflict, + FragmentAdapterReadError, + FragmentAdapterRegistry, + FragmentAdapterRemoved, + FragmentAdapterState, + InMemoryFragmentAdapterStateStore, +) +from tests.test_lattice_autoconfig import snapshot # noqa: E402 + + +class FragmentComponent: + """Brand-neutral component exposing only the common discovery method.""" + + def __init__(self, adapter): + """Store the adapter returned during discovery.""" + self.adapter = adapter + self.calls = 0 + + def lattice_fragment_adapter(self): + """Return the component-owned fragment publisher.""" + self.calls += 1 + return self.adapter + + +class UnrelatedComponent: + """Component without any Lattice fragment publisher surface.""" + + +class RaisingLoadStore(InMemoryFragmentAdapterStateStore): + """State store whose durable read is unavailable.""" + + def load(self): + """Raise one representative durable-store fault.""" + raise OSError("disk unavailable") + + +class RejectingStore(InMemoryFragmentAdapterStateStore): + """State store rejecting every atomic fragment write.""" + + def compare_and_store(self, expected, replacement): + """Reject the candidate without changing durable state.""" + return False + + +class ToggleLoadStore(InMemoryFragmentAdapterStateStore): + """State store that can fail after an initial successful compilation.""" + + def __init__(self): + """Create an initially available durable store.""" + super().__init__() + self.fail_reads = False + + def load(self): + """Return state until the test makes durable reads unavailable.""" + if self.fail_reads: + raise OSError("durable fragment unavailable") + return super().load() + + +def publisher(provider_id, generation=1, health=ProviderHealth.HEALTHY): + """Build one seeded durable generic publisher.""" + state_store = InMemoryFragmentAdapterStateStore() + adapter = DurableFragmentAdapter(provider_id, state_store) + initial = snapshot( + provider_id, + generation=generation, + node_id="{}-INV".format(provider_id.upper()), + health=health, + ) + adapter.publish(initial, "initial discovery") + return adapter, state_store, initial + + +def advance( + current, + generation, + health=None, + node_id=None, +): + """Build the next immutable test snapshot without copying frozen mappings.""" + if health is None: + health = current.health + if node_id is None: + node_id = current.topology_fragment["nodes"][0]["id"] + return snapshot( + current.provider_id, + generation=generation, + node_id=node_id, + health=health, + ) + + +def compiled_registry(*adapters): + """Discover generic components and create a durable compiler.""" + registry = FragmentAdapterRegistry(enabled=True) + components = [ + UnrelatedComponent(), + ] + [FragmentComponent(adapter) for adapter in adapters] + discovered = registry.discover(components) + compiled_store = InMemoryCompiledLatticeStateStore() + compiler = registry.create_compiler(compiled_store) + return registry, compiler, compiled_store, discovered + + +class TestDurableFragmentAdapter(unittest.TestCase): + """Each integration owns a monotonic durable immutable fragment cursor.""" + + def test_seed_and_fresh_reads_are_immutable_and_durable(self): + """A published snapshot is detached and restored from its store.""" + adapter, state_store, initial = publisher("gateway") + + state = adapter.read_state() + + self.assertIsInstance(state, FragmentAdapterState) + self.assertEqual(state.generation, 1) + self.assertEqual(len(state.semantic_fingerprint), 64) + self.assertIs(state.snapshot, initial) + self.assertIs(adapter.read_snapshot(), initial) + self.assertEqual(state_store.writes, 1) + + restarted = DurableFragmentAdapter("gateway", state_store) + self.assertEqual(restarted.read_state(), state) + self.assertIs(restarted.read_snapshot(), initial) + + def test_restart_rejects_regression_and_generation_reuse(self): + """Durable cursor restoration rejects regressions and mutations.""" + _adapter, state_store, initial = publisher("cloud", generation=7) + restarted = DurableFragmentAdapter("cloud", state_store) + + with self.assertRaisesRegex(ValueError, "regressed"): + restarted.publish( + advance(initial, 6), + "stale cache", + ) + with self.assertRaisesRegex(ValueError, "reused"): + restarted.publish( + advance(initial, 7, node_id="MUTATED"), + "same generation mutation", + ) + self.assertFalse( + restarted.publish(initial, "exact replay"), + ) + self.assertEqual(state_store.writes, 1) + + def test_reader_store_failure_is_wrapped_and_fails_closed(self): + """No synthetic or empty fragment is returned on durable read failure.""" + store = RaisingLoadStore() + + with self.assertRaisesRegex( + FragmentAdapterReadError, + "disk unavailable", + ): + DurableFragmentAdapter("gateway", store) + + def test_conflicting_atomic_write_leaves_requested_generation_pending(self): + """A rejected CAS never presents an uncommitted fragment as current.""" + adapter, seeded_store, initial = publisher("gateway") + rejecting = RejectingStore(adapter.read_state()) + adapter = DurableFragmentAdapter("gateway", rejecting) + _registry, compiler, _compiled_store, _discovered = compiled_registry(adapter) + first = compiler.drain() + self.assertEqual(first.status, CompileStatus.FRESH) + + updated = advance(initial, 2) + with self.assertRaises(FragmentAdapterConflict): + adapter.publish(updated, "new telemetry") + + failed = compiler.drain() + self.assertEqual(failed.status, CompileStatus.STALE) + self.assertTrue(failed.pending) + self.assertEqual(adapter.read_state(), seeded_store.load()) + self.assertEqual( + first.publication, + compiler.publication, + ) + + def test_removal_is_durable_and_reader_fails_closed_after_restart(self): + """Runtime removal is an auditable tombstone, never silent absence.""" + adapter, state_store, _initial = publisher("gateway") + + self.assertTrue(adapter.remove(2, "integration removed")) + removed = adapter.read_state() + + self.assertTrue(removed.removed) + self.assertEqual(removed.generation, 2) + self.assertEqual(removed.snapshot.health, ProviderHealth.OFFLINE) + with self.assertRaises(FragmentAdapterRemoved): + adapter.read_snapshot() + + restarted = DurableFragmentAdapter("gateway", state_store) + self.assertEqual(restarted.read_state(), removed) + with self.assertRaises(FragmentAdapterRemoved): + restarted.read_snapshot() + + +class TestFragmentAdapterRegistry(unittest.TestCase): + """Any common-surface component can drive the compiled coordinator.""" + + def test_registry_is_default_off_and_does_not_touch_components(self): + """Disabled discovery performs no integration calls or registration.""" + adapter, _store, _initial = publisher("gateway") + component = FragmentComponent(adapter) + registry = FragmentAdapterRegistry() + + self.assertEqual(registry.discover([component]), ()) + self.assertEqual(component.calls, 0) + self.assertEqual(registry.provider_ids, ()) + self.assertFalse(registry.register(adapter)) + self.assertFalse(registry.unregister("gateway")) + with self.assertRaisesRegex(RuntimeError, "disabled"): + registry.create_compiler(InMemoryCompiledLatticeStateStore()) + + def test_discovery_has_no_brand_allow_list(self): + """Arbitrary provider IDs are discovered through the common surface.""" + alpha, _alpha_store, _alpha = publisher("future-cloud-alpha") + beta, _beta_store, _beta = publisher("local-modbus-beta") + + registry, compiler, _store, discovered = compiled_registry( + alpha, + beta, + ) + run = compiler.drain() + + self.assertEqual( + discovered, + ("future-cloud-alpha", "local-modbus-beta"), + ) + self.assertEqual( + registry.provider_ids, + ("future-cloud-alpha", "local-modbus-beta"), + ) + self.assertEqual( + dict(run.publication.provider_generations), + { + "future-cloud-alpha": 1, + "local-modbus-beta": 1, + }, + ) + + def test_invalid_discovery_batch_is_transactional(self): + """Duplicate discovery does not partially mutate registry membership.""" + alpha, _alpha_store, _initial = publisher("same-provider") + duplicate = DurableFragmentAdapter( + "same-provider", + _alpha_store, + ) + registry = FragmentAdapterRegistry(enabled=True) + + with self.assertRaisesRegex(ValueError, "more than once"): + registry.discover( + [ + FragmentComponent(alpha), + FragmentComponent(duplicate), + ] + ) + + self.assertEqual(registry.provider_ids, ()) + + def test_register_unregister_only_before_compiler_is_sealed(self): + """Runtime membership changes cannot silently alter compiler inputs.""" + adapter, _store, _initial = publisher("gateway") + registry = FragmentAdapterRegistry(enabled=True) + + self.assertTrue(registry.register(adapter)) + self.assertTrue(registry.unregister("gateway")) + self.assertTrue(registry.register(adapter)) + compiler = registry.create_compiler(InMemoryCompiledLatticeStateStore()) + + with self.assertRaisesRegex(RuntimeError, "tombstone"): + registry.unregister("gateway") + with self.assertRaisesRegex(RuntimeError, "sealed"): + registry.register(adapter) + self.assertEqual(compiler.drain().status, CompileStatus.FRESH) + + def test_any_registered_integration_invalidates_and_republishes(self): + """Every discovered publisher can replace its prior fragment.""" + alpha, _alpha_store, alpha_one = publisher("alpha") + beta, _beta_store, beta_one = publisher("beta") + _registry, compiler, compiled_store, _ids = compiled_registry( + alpha, + beta, + ) + first = compiler.drain() + + alpha_two = advance(alpha_one, 2) + beta_two = advance(beta_one, 2) + self.assertTrue(alpha.publish(alpha_two, "alpha refresh")) + self.assertTrue(beta.publish(beta_two, "beta refresh")) + second = compiler.drain() + + self.assertEqual(second.attempts, 1) + self.assertTrue(second.published) + self.assertEqual(second.publication.lattice_version, 2) + self.assertEqual( + dict(second.publication.provider_generations), + {"alpha": 2, "beta": 2}, + ) + self.assertEqual(compiled_store.writes, 2) + self.assertEqual( + {(cause.source_id, cause.generation) for cause in second.publication.invalidation_causes}, + {("alpha", 2), ("beta", 2)}, + ) + self.assertEqual(first.publication.lattice_version, 1) + + def test_degraded_fragment_invalidates_and_publishes_degraded_cursor(self): + """Health-only changes are durable inputs and trigger recompilation.""" + adapter, _store, initial = publisher("gateway") + _registry, compiler, _compiled_store, _ids = compiled_registry(adapter) + compiler.drain() + + degraded = advance( + initial, + 2, + health=ProviderHealth.DEGRADED, + ) + self.assertTrue(adapter.publish(degraded, "provider health degraded")) + run = compiler.drain() + + self.assertTrue(run.published) + self.assertEqual(run.status, CompileStatus.DEGRADED) + self.assertEqual( + dict(run.publication.provider_generations), + {"gateway": 2}, + ) + self.assertIn( + "provider_degraded", + {issue.code for issue in run.issues}, + ) + + def test_offline_fragment_invalidates_but_preserves_last_known_good(self): + """An active provider going offline fails closed after one attempt.""" + adapter, _store, initial = publisher("gateway") + _registry, compiler, _compiled_store, _ids = compiled_registry(adapter) + first = compiler.drain() + + offline = advance( + initial, + 2, + health=ProviderHealth.OFFLINE, + ) + self.assertTrue(adapter.publish(offline, "provider disconnected")) + run = compiler.drain() + + self.assertEqual(run.attempts, 1) + self.assertEqual(run.status, CompileStatus.STALE) + self.assertTrue(run.pending) + self.assertIs(run.publication, first.publication) + self.assertIs(run.plan, first.publication.plan) + self.assertIn( + "active_provider_unavailable", + {issue.code for issue in run.issues}, + ) + + def test_other_provider_invalidation_exposes_reader_failure_fail_closed(self): + """A fresh-read failure blocks publication instead of dropping input.""" + gateway_store = ToggleLoadStore() + gateway = DurableFragmentAdapter("gateway", gateway_store) + gateway_one = snapshot("gateway", generation=1, node_id="GW-INV") + gateway.publish(gateway_one, "initial gateway discovery") + cloud, _cloud_store, cloud_one = publisher("cloud") + _registry, compiler, _compiled_store, _ids = compiled_registry( + gateway, + cloud, + ) + first = compiler.drain() + + gateway_store.fail_reads = True + self.assertTrue( + cloud.publish( + advance(cloud_one, 2), + "cloud refresh requires fresh read of every provider", + ) + ) + run = compiler.drain() + + self.assertEqual(run.attempts, 1) + self.assertEqual(run.status, CompileStatus.STALE) + self.assertTrue(run.pending) + self.assertIs(run.publication, first.publication) + self.assertIn( + "provider_read_failed", + {issue.code for issue in run.issues}, + ) + + def test_removal_invalidates_and_preserves_last_known_good(self): + """A durable removal triggers a bounded fail-closed recompile.""" + adapter, _store, _initial = publisher("gateway") + _registry, compiler, _compiled_store, _ids = compiled_registry(adapter) + first = compiler.drain() + + self.assertTrue(adapter.remove(2, "integration disabled by user")) + run = compiler.drain() + + self.assertEqual(run.attempts, 1) + self.assertEqual(run.status, CompileStatus.STALE) + self.assertTrue(run.pending) + self.assertIs(run.publication, first.publication) + self.assertIn( + "provider_read_failed", + {issue.code for issue in run.issues}, + ) + + def test_restart_restores_adapter_and_compiler_cursor_protection(self): + """Both durable layers reject reuse after a complete process restart.""" + adapter, adapter_store, initial = publisher("gateway") + registry, compiler, compiled_store, _ids = compiled_registry(adapter) + compiler.drain() + second_snapshot = advance(initial, 2) + self.assertTrue(adapter.publish(second_snapshot, "new discovery")) + second = compiler.drain() + self.assertEqual(second.publication.lattice_version, 2) + + restarted_adapter = DurableFragmentAdapter( + "gateway", + adapter_store, + ) + restarted_registry, restarted_compiler, _same_store, _ids = compiled_registry_with_store( + compiled_store, + restarted_adapter, + ) + exact = restarted_compiler.drain() + + self.assertEqual( + restarted_registry.provider_ids, + registry.provider_ids, + ) + self.assertFalse(exact.published) + self.assertEqual(exact.publication, second.publication) + with self.assertRaisesRegex(ValueError, "reused"): + restarted_adapter.publish( + advance(second_snapshot, 2, node_id="REUSED"), + "same cursor mutation after restart", + ) + self.assertEqual( + restarted_compiler.publication, + second.publication, + ) + + def test_publication_feedback_token_does_not_persist_or_recompile(self): + """Compiler-origin feedback is suppressed before adapter persistence.""" + adapter, state_store, initial = publisher("gateway") + _registry, compiler, compiled_store, _ids = compiled_registry(adapter) + first = compiler.drain() + + feedback_snapshot = advance(initial, 2) + self.assertFalse( + adapter.publish( + feedback_snapshot, + "published config observed", + feedback_token=first.publication.feedback_token, + ) + ) + idle = compiler.drain() + + self.assertEqual(adapter.read_state().generation, 1) + self.assertEqual(state_store.writes, 1) + self.assertEqual(compiled_store.writes, 1) + self.assertEqual(idle.attempts, 0) + self.assertIs(idle.publication, first.publication) + + def test_invalidation_during_read_gets_one_bounded_follow_up(self): + """Concurrent adapter invalidation triggers exactly one fresh follow-up.""" + adapter, _store, initial = publisher("gateway") + registry = FragmentAdapterRegistry(enabled=True) + registry.register(adapter) + original_reader = adapter.read_snapshot + fired = [False] + + def invalidating_reader(): + """Publish one newer generation during the first compile read.""" + value = original_reader() + if not fired[0]: + fired[0] = True + adapter.publish( + advance(initial, 2), + "concurrent rediscovery", + ) + return value + + adapter.read_snapshot = invalidating_reader + compiler = registry.create_compiler(InMemoryCompiledLatticeStateStore()) + run = compiler.drain() + + self.assertEqual(run.attempts, 2) + self.assertTrue(run.published) + self.assertEqual( + dict(run.publication.provider_generations), + {"gateway": 2}, + ) + + +def compiled_registry_with_store(compiled_store, *adapters): + """Create a restarted registry against an existing compiled store.""" + registry = FragmentAdapterRegistry(enabled=True) + discovered = registry.discover([FragmentComponent(adapter) for adapter in adapters]) + compiler = registry.create_compiler(compiled_store) + return registry, compiler, compiled_store, discovered + + +if __name__ == "__main__": + unittest.main()