From b9d782fdd4816d9563e660b86bed3a21aef89d3e Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Mon, 27 Jul 2026 23:03:26 +0100 Subject: [PATCH] feat: compile indexed lattice roles --- apps/predbat/lattice_autoconfig.py | 310 +++++++++++++++- apps/predbat/tests/test_lattice_autoconfig.py | 349 ++++++++++++++++-- 2 files changed, 607 insertions(+), 52 deletions(-) diff --git a/apps/predbat/lattice_autoconfig.py b/apps/predbat/lattice_autoconfig.py index 66922793e..847ff438f 100644 --- a/apps/predbat/lattice_autoconfig.py +++ b/apps/predbat/lattice_autoconfig.py @@ -98,6 +98,29 @@ def __post_init__(self): object.__setattr__(self, "node_id", self.node_id.strip()) +@dataclass(frozen=True) +class ProviderRoleAssignment: + """A provider-local indexed primary or control role assertion.""" + + role: AliasRole + group: str + index: int + node_id: str + + def __post_init__(self): + """Validate and normalise one indexed role assertion.""" + if self.role not in (AliasRole.PRIMARY, AliasRole.CONTROL): + raise ValueError("role assignment must be PRIMARY or CONTROL") + if not isinstance(self.group, str) or not self.group.strip(): + raise ValueError("role assignment group must be a non-empty string") + if not isinstance(self.index, int) or isinstance(self.index, bool) or self.index < 0: + raise ValueError("role assignment index must be a non-negative integer") + if not isinstance(self.node_id, str) or not self.node_id.strip(): + raise ValueError("role assignment node_id must be a non-empty string") + object.__setattr__(self, "group", self.group.strip()) + object.__setattr__(self, "node_id", self.node_id.strip()) + + @dataclass(frozen=True) class ProviderSnapshot: """One integration's immutable generation of health, topology, and aliases.""" @@ -108,6 +131,7 @@ class ProviderSnapshot: topology_fragment: Mapping aliases: tuple = () identity_aliases: tuple = () + role_assignments: tuple = () def __post_init__(self): """Validate scalar fields and detach caller-owned mutable data.""" @@ -125,10 +149,14 @@ def __post_init__(self): identity_aliases = tuple(self.identity_aliases) if any(not isinstance(alias, ProviderIdentityAlias) for alias in identity_aliases): raise ValueError("identity_aliases must contain ProviderIdentityAlias values") + role_assignments = tuple(self.role_assignments) + if any(not isinstance(assignment, ProviderRoleAssignment) for assignment in role_assignments): + raise ValueError("role_assignments must contain ProviderRoleAssignment values") object.__setattr__(self, "provider_id", self.provider_id.strip()) object.__setattr__(self, "topology_fragment", _freeze(copy.deepcopy(dict(self.topology_fragment)))) object.__setattr__(self, "aliases", aliases) object.__setattr__(self, "identity_aliases", identity_aliases) + object.__setattr__(self, "role_assignments", role_assignments) @dataclass(frozen=True) @@ -154,6 +182,58 @@ class IdentityBinding: canonical_node_id: str +@dataclass(frozen=True) +class RoleAssignmentBinding: + """One normalized provider assertion for an indexed plan role.""" + + provider_id: str + generation: int + role: str + group: str + index: int + local_node_id: str + canonical_node_id: str + + +@dataclass(frozen=True) +class IndexedRoleTarget: + """One deterministic indexed primary or control target.""" + + group: str + index: int + node_id: str + provenance: tuple + + +@dataclass(frozen=True) +class MaterializationReadiness: + """Fail-closed decision exposed to a future config materializer.""" + + ready: bool + blockers: tuple + + def __post_init__(self): + """Normalise blockers and reject contradictory readiness state.""" + if not isinstance(self.ready, bool): + raise ValueError("materialization readiness must be a boolean") + if isinstance(self.blockers, str): + raise ValueError("materialization blockers must be an iterable of strings") + try: + blockers = tuple(self.blockers) + except TypeError as exc: + raise ValueError("materialization blockers must be an iterable of strings") from exc + if any(not isinstance(blocker, str) for blocker in blockers): + raise ValueError("materialization blockers must be non-empty strings") + blockers = tuple(blocker.strip() for blocker in blockers) + if any(not blocker for blocker in blockers): + raise ValueError("materialization blockers must be non-empty strings") + if len(blockers) != len(set(blockers)): + raise ValueError("materialization blockers must be unique") + if self.ready != (not blockers): + raise ValueError("materialization readiness must equal absence of blockers") + object.__setattr__(self, "blockers", blockers) + + @dataclass(frozen=True) class FieldProvenance: """Source coordinates for one generated plan field.""" @@ -181,6 +261,12 @@ class AutoConfigPlan: topology: Mapping aliases: tuple identity_aliases: tuple + role_assignments: tuple + primary_targets: tuple + control_targets: tuple + primary_target: Optional[str] + control_target: Optional[str] + materialization_readiness: MaterializationReadiness fields: tuple provenance: tuple provider_generations: tuple @@ -282,6 +368,15 @@ def _fingerprint_snapshot(snapshot): } for alias in sorted(snapshot.identity_aliases, key=lambda item: (item.kind, item.value, item.node_id)) ] + role_assignments = [ + { + "role": assignment.role.value, + "group": assignment.group, + "index": assignment.index, + "node_id": assignment.node_id, + } + for assignment in sorted(snapshot.role_assignments, key=lambda item: (item.group, item.role.value, item.index, item.node_id)) + ] payload = { "provider": snapshot.provider_id, "generation": snapshot.generation, @@ -289,6 +384,7 @@ def _fingerprint_snapshot(snapshot): "fragment": snapshot.topology_fragment, "aliases": aliases, "identity_aliases": identity_aliases, + "role_assignments": role_assignments, } return hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest() @@ -449,7 +545,154 @@ def _correlate_identities(snapshots, documents, provider_nodes): return tuple(normalized_documents), canonical_by_node, identity_bindings -def _field_provenance(snapshots, bindings, identity_bindings, primary_target, control_target, topology_snapshot, canonical_by_node): +def _compile_roles(snapshots, bindings, identity_bindings, provider_nodes, canonical_by_node): + """Validate legacy/indexed roles and return deterministic target outputs.""" + legacy_by_role = {role: tuple(binding for binding in bindings if role.value in binding.roles) for role in (AliasRole.PRIMARY, AliasRole.CONTROL)} + legacy_assignments = tuple(binding for role_bindings in legacy_by_role.values() for binding in role_bindings) + + role_bindings = [] + qualified_assignments = set() + generation_by_provider = {snapshot.provider_id: snapshot.generation for snapshot in snapshots} + for snapshot in snapshots: + assignments = sorted(snapshot.role_assignments, key=lambda item: (item.group, item.role.value, item.index, item.node_id)) + for assignment in assignments: + if assignment.node_id not in provider_nodes[snapshot.provider_id]: + raise AutoConfigCompileError( + "role assignment {}:{}:{} targets unknown provider-local node {}".format( + assignment.group, + assignment.role.value, + assignment.index, + assignment.node_id, + ) + ) + qualified_key = (snapshot.provider_id, assignment.group, assignment.role.value, assignment.index) + if qualified_key in qualified_assignments: + raise AutoConfigCompileError( + "role assignment collision for {}:{}:{}:{}".format( + snapshot.provider_id, + assignment.group, + assignment.role.value, + assignment.index, + ) + ) + qualified_assignments.add(qualified_key) + role_bindings.append( + RoleAssignmentBinding( + provider_id=snapshot.provider_id, + generation=snapshot.generation, + role=assignment.role.value, + group=assignment.group, + index=assignment.index, + local_node_id=assignment.node_id, + canonical_node_id=canonical_by_node[(snapshot.provider_id, assignment.node_id)], + ) + ) + role_bindings = tuple( + sorted( + role_bindings, + key=lambda item: (item.group, item.role, item.index, item.provider_id, item.local_node_id), + ) + ) + + if role_bindings and legacy_assignments: + raise AutoConfigCompileError("legacy and indexed role assignments cannot be mixed") + + legacy_targets = {} + if not role_bindings: + for role, role_aliases in legacy_by_role.items(): + if len(role_aliases) > 1: + raise AutoConfigCompileError("ambiguous legacy {} assignments".format(role.value)) + legacy_targets[role] = role_aliases[0].node_id if role_aliases else None + else: + legacy_targets = {AliasRole.PRIMARY: None, AliasRole.CONTROL: None} + + indices_by_group_role = {} + assignments_by_slot = {} + for binding in role_bindings: + group_role = (binding.group, binding.role) + indices_by_group_role.setdefault(group_role, set()).add(binding.index) + assignments_by_slot.setdefault((binding.group, binding.role, binding.index), []).append(binding) + for (group, role), indices in sorted(indices_by_group_role.items()): + ordered = sorted(indices) + expected = list(range(ordered[-1] + 1)) + if ordered != expected: + raise AutoConfigCompileError( + "{} {} indices must be contiguous from zero; got {}".format( + group, + role, + ordered, + ) + ) + + correlated_providers = {} + for binding in identity_bindings: + correlated_providers.setdefault(binding.canonical_node_id, set()).add(binding.provider_id) + + indexed_targets = {AliasRole.PRIMARY: [], AliasRole.CONTROL: []} + for (group, role_value, index), assignments in sorted(assignments_by_slot.items()): + canonical_nodes = {assignment.canonical_node_id for assignment in assignments} + if len(canonical_nodes) != 1: + raise AutoConfigCompileError( + "conflicting {} target for {} index {}: {}".format( + role_value, + group, + index, + sorted(canonical_nodes), + ) + ) + canonical_node_id = next(iter(canonical_nodes)) + providers = {assignment.provider_id for assignment in assignments} + if len(providers) > 1 and not providers.issubset(correlated_providers.get(canonical_node_id, set())): + raise AutoConfigCompileError( + "providers sharing {} {} index {} require explicit strong identity correlation".format( + group, + role_value, + index, + ) + ) + provenance = tuple( + FieldProvenance( + "/{}_targets/{}/{}/node_id".format(role_value, group, index), + assignment.provider_id, + generation_by_provider[assignment.provider_id], + "/role_assignments/{}/{}/{}".format(role_value, group, index), + ) + for assignment in assignments + ) + indexed_targets[AliasRole(role_value)].append( + IndexedRoleTarget( + group=group, + index=index, + node_id=canonical_node_id, + provenance=provenance, + ) + ) + + primary_targets = tuple(indexed_targets[AliasRole.PRIMARY]) + control_targets = tuple(indexed_targets[AliasRole.CONTROL]) + blockers = [] + if not primary_targets: + blockers.append("indexed_primary_targets_missing") + if not control_targets: + blockers.append("indexed_control_targets_missing") + if legacy_assignments: + blockers.append("legacy_role_assignments_present") + blockers.append("config_projection_bindings_missing") + readiness = MaterializationReadiness(ready=False, blockers=tuple(blockers)) + return role_bindings, primary_targets, control_targets, legacy_targets, readiness + + +def _field_provenance( + snapshots, + bindings, + identity_bindings, + primary_target, + control_target, + primary_targets, + control_targets, + topology_snapshot, + canonical_by_node, +): """Build deterministic source coordinates for every generated field.""" provenance = [] for snapshot in snapshots: @@ -485,6 +728,8 @@ def _field_provenance(snapshots, bindings, identity_bindings, primary_target, co "/aliases/{}".format(binding.qualified_name.split(":", 1)[1]), ) ) + for target in primary_targets + control_targets: + provenance.extend(target.provenance) generations = {snapshot.provider_id: snapshot.generation for snapshot in snapshots} local_by_canonical = {(provider_id, canonical_node_id): local_node_id for (provider_id, local_node_id), canonical_node_id in canonical_by_node.items()} for key, source in topology_snapshot.provenance.items(): @@ -544,12 +789,15 @@ def compile_auto_config(snapshots): ) bindings = tuple(sorted(bindings, key=lambda item: item.qualified_name)) - targets = {} - for role in (AliasRole.PRIMARY, AliasRole.CONTROL): - role_targets = sorted({binding.node_id for binding in bindings if role.value in binding.roles}) - if len(role_targets) > 1: - raise AutoConfigCompileError("ambiguous {} target: {}".format(role.value, ", ".join(role_targets))) - targets[role] = role_targets[0] if role_targets else None + role_bindings, primary_targets, control_targets, legacy_targets, readiness = _compile_roles( + snapshots, + bindings, + identity_bindings, + provider_nodes, + canonical_by_node, + ) + primary_target = legacy_targets[AliasRole.PRIMARY] + control_target = legacy_targets[AliasRole.CONTROL] topology_snapshot = merge_topologies(documents) fields = [] @@ -561,8 +809,10 @@ def compile_auto_config(snapshots): "/aliases/{}".format(binding.qualified_name.split(":", 1)[1]), ) fields.append(AutoConfigField("alias.{}".format(binding.qualified_name), binding.node_id, (source,))) - for name, role in (("primary_target", AliasRole.PRIMARY), ("control_target", AliasRole.CONTROL)): - target = targets[role] + for name, target, role in ( + ("primary_target", primary_target, AliasRole.PRIMARY), + ("control_target", control_target, AliasRole.CONTROL), + ): if target is None: continue sources = tuple( @@ -576,6 +826,15 @@ def compile_auto_config(snapshots): if binding.node_id == target and role.value in binding.roles ) fields.append(AutoConfigField(name, target, sources)) + for name, targets in (("primary_targets", primary_targets), ("control_targets", control_targets)): + for target in targets: + fields.append( + AutoConfigField( + "{}.{}.{}".format(name, target.group, target.index), + target.node_id, + target.provenance, + ) + ) fields = tuple(sorted(fields, key=lambda item: item.name)) semantic = { @@ -598,6 +857,25 @@ def compile_auto_config(snapshots): } for binding in identity_bindings ], + "role_assignments": [ + { + "provider_id": binding.provider_id, + "role": binding.role, + "group": binding.group, + "index": binding.index, + "local_node_id": binding.local_node_id, + "canonical_node_id": binding.canonical_node_id, + } + for binding in role_bindings + ], + "primary_targets": [{"group": target.group, "index": target.index, "node_id": target.node_id} for target in primary_targets], + "control_targets": [{"group": target.group, "index": target.index, "node_id": target.node_id} for target in control_targets], + "primary_target": primary_target, + "control_target": control_target, + "materialization_readiness": { + "ready": readiness.ready, + "blockers": readiness.blockers, + }, "fields": [{"name": field.name, "value": field.value} for field in fields], } digest = hashlib.sha256(_canonical_json(semantic).encode("utf-8")).hexdigest() @@ -605,8 +883,10 @@ def compile_auto_config(snapshots): snapshots, bindings, identity_bindings, - targets[AliasRole.PRIMARY], - targets[AliasRole.CONTROL], + primary_target, + control_target, + primary_targets, + control_targets, topology_snapshot, canonical_by_node, ) @@ -615,6 +895,12 @@ def compile_auto_config(snapshots): topology=_freeze(topology_snapshot.site), aliases=bindings, identity_aliases=identity_bindings, + role_assignments=role_bindings, + primary_targets=primary_targets, + control_targets=control_targets, + primary_target=primary_target, + control_target=control_target, + materialization_readiness=readiness, fields=fields, provenance=provenance, provider_generations=tuple((snapshot.provider_id, snapshot.generation) for snapshot in snapshots), @@ -758,7 +1044,7 @@ def _compile_attempt(self): def _materialize_if_changed(self, plan): """Hand a changed plan to the injected materializer exactly once.""" with self._lock: - if self._materializer is None or (self._active_plan is not None and plan.digest == self._active_plan.digest): + if not plan.materialization_readiness.ready or self._materializer is None or (self._active_plan is not None and plan.digest == self._active_plan.digest): return 0, () self._token_counter += 1 feedback_token = "lattice-autoconfig-{}".format(self._token_counter) diff --git a/apps/predbat/tests/test_lattice_autoconfig.py b/apps/predbat/tests/test_lattice_autoconfig.py index 5a1d14aab..285fb4010 100644 --- a/apps/predbat/tests/test_lattice_autoconfig.py +++ b/apps/predbat/tests/test_lattice_autoconfig.py @@ -6,6 +6,7 @@ import sys import threading import unittest +from dataclasses import replace sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) @@ -14,9 +15,11 @@ AutoConfigCompileError, CompileStatus, LatticeAutoConfigCompiler, + MaterializationReadiness, ProviderAlias, ProviderHealth, ProviderIdentityAlias, + ProviderRoleAssignment, ProviderSnapshot, compile_auto_config, ) @@ -50,9 +53,35 @@ def fragment(provider, generation, node_id="INV1", kind="inverter"): } -def snapshot(provider, generation=1, node_id="INV1", kind="inverter", health=ProviderHealth.HEALTHY, aliases=(), identity_aliases=()): +def snapshot( + provider, + generation=1, + node_id="INV1", + kind="inverter", + health=ProviderHealth.HEALTHY, + aliases=(), + identity_aliases=(), + role_assignments=(), +): """Build one typed provider snapshot.""" - return ProviderSnapshot(provider, generation, health, fragment(provider, generation, node_id=node_id, kind=kind), aliases, identity_aliases) + return ProviderSnapshot( + provider, + generation, + health, + fragment(provider, generation, node_id=node_id, kind=kind), + aliases, + identity_aliases, + role_assignments, + ) + + +def ready_plan(): + """Copy a compiled shadow plan into a future-materializer test harness.""" + plan = compile_auto_config((snapshot("gateway"),)) + return replace( + plan, + materialization_readiness=MaterializationReadiness(True, ()), + ) class MutableReader: @@ -72,11 +101,34 @@ def __call__(self): class TestPlanCompilation(unittest.TestCase): """Compiler output is safe, immutable, deterministic, and attributable.""" + def test_materialization_readiness_rejects_inconsistent_state(self): + """Readiness is derived exactly from a normalized blocker tuple.""" + readiness = MaterializationReadiness( + False, + [" config_projection_bindings_missing "], + ) + + self.assertEqual( + readiness.blockers, + ("config_projection_bindings_missing",), + ) + with self.assertRaisesRegex(ValueError, "absence of blockers"): + MaterializationReadiness(True, ("projection_missing",)) + with self.assertRaisesRegex(ValueError, "absence of blockers"): + MaterializationReadiness(False, ()) + with self.assertRaisesRegex(ValueError, "unique"): + MaterializationReadiness(False, ("projection_missing", "projection_missing")) + with self.assertRaisesRegex(ValueError, "non-empty"): + MaterializationReadiness(False, (" ",)) + with self.assertRaisesRegex(ValueError, "iterable of strings"): + MaterializationReadiness(False, "projection_missing") + def test_order_independent_digest_and_provider_qualified_aliases(self): """Input order and shared local alias names cannot alter a plan.""" - alias = ProviderAlias("battery", "INV1", frozenset((AliasRole.REFERENCE, AliasRole.PRIMARY, AliasRole.CONTROL))) - gateway = snapshot("gateway", aliases=(alias,), identity_aliases=(ProviderIdentityAlias("serial", "SER123", "INV1"),)) - cloud = snapshot("cloud", aliases=(alias,), identity_aliases=(ProviderIdentityAlias("serial", "SER123", "INV1"),)) + gateway_alias = ProviderAlias("battery", "INV1", frozenset((AliasRole.REFERENCE, AliasRole.PRIMARY, AliasRole.CONTROL))) + cloud_alias = ProviderAlias("battery", "INV1") + gateway = snapshot("gateway", aliases=(gateway_alias,), identity_aliases=(ProviderIdentityAlias("serial", "SER123", "INV1"),)) + cloud = snapshot("cloud", aliases=(cloud_alias,), identity_aliases=(ProviderIdentityAlias("serial", "SER123", "INV1"),)) left = compile_auto_config((gateway, cloud)) right = compile_auto_config((cloud, gateway)) @@ -85,6 +137,10 @@ def test_order_independent_digest_and_provider_qualified_aliases(self): self.assertEqual([binding.qualified_name for binding in left.aliases], ["cloud:battery", "gateway:battery"]) self.assertEqual(dict(left.provider_generations), {"cloud": 1, "gateway": 1}) self.assertEqual({field.name for field in left.fields}, {"alias.cloud:battery", "alias.gateway:battery", "control_target", "primary_target"}) + self.assertEqual(left.primary_target, "identity:serial:SER123") + self.assertEqual(left.control_target, "identity:serial:SER123") + self.assertEqual(left.primary_targets, ()) + self.assertEqual(left.control_targets, ()) self.assertTrue(all(field.provenance for field in left.fields)) with self.assertRaises(TypeError): left.topology["scope"] = "fragment" @@ -181,12 +237,12 @@ def test_ambiguous_primary_and_control_targets_fail_closed(self): """Several distinct target nodes cannot silently pick a winner.""" primary_a = ProviderAlias("battery", "INV1", frozenset((AliasRole.PRIMARY,))) primary_b = ProviderAlias("battery", "INV2", frozenset((AliasRole.PRIMARY,))) - with self.assertRaisesRegex(AutoConfigCompileError, "ambiguous primary"): + with self.assertRaisesRegex(AutoConfigCompileError, "ambiguous legacy primary"): compile_auto_config((snapshot("gateway", node_id="INV1", aliases=(primary_a,)), snapshot("cloud", node_id="INV2", aliases=(primary_b,)))) control_a = ProviderAlias("control", "INV1", frozenset((AliasRole.CONTROL,))) control_b = ProviderAlias("control", "INV2", frozenset((AliasRole.CONTROL,))) - with self.assertRaisesRegex(AutoConfigCompileError, "ambiguous control"): + with self.assertRaisesRegex(AutoConfigCompileError, "ambiguous legacy control"): compile_auto_config((snapshot("gateway", node_id="INV1", aliases=(control_a,)), snapshot("cloud", node_id="INV2", aliases=(control_b,)))) def test_alias_must_target_provider_local_identity(self): @@ -195,6 +251,157 @@ def test_alias_must_target_provider_local_identity(self): with self.assertRaisesRegex(AutoConfigCompileError, "unknown provider-local node"): compile_auto_config((snapshot("gateway", node_id="INV1", aliases=(bad,)), snapshot("cloud", node_id="INV2"))) + def test_indexed_roles_are_order_independent_and_contiguous(self): + """Indexed targets sort by group, role, and index regardless of input order.""" + first = snapshot( + "a", + node_id="A", + role_assignments=( + ProviderRoleAssignment(AliasRole.CONTROL, "battery", 1, "A"), + ProviderRoleAssignment(AliasRole.PRIMARY, "battery", 0, "A"), + ), + ) + second = snapshot( + "z", + node_id="Z", + role_assignments=( + ProviderRoleAssignment(AliasRole.PRIMARY, "battery", 1, "Z"), + ProviderRoleAssignment(AliasRole.CONTROL, "battery", 0, "Z"), + ), + ) + + left = compile_auto_config((second, first)) + right = compile_auto_config((first, second)) + + self.assertEqual(left.digest, right.digest) + self.assertEqual( + [(target.group, target.index, target.node_id) for target in left.primary_targets], + [ + ("battery", 0, "provider:a:A"), + ("battery", 1, "provider:z:Z"), + ], + ) + self.assertEqual( + [(target.group, target.index, target.node_id) for target in left.control_targets], + [ + ("battery", 0, "provider:z:Z"), + ("battery", 1, "provider:a:A"), + ], + ) + self.assertEqual( + [(item.group, item.role, item.index) for item in left.role_assignments], + sorted((item.group, item.role, item.index) for item in left.role_assignments), + ) + self.assertEqual( + {field.name for field in left.fields if field.name.startswith(("primary_targets", "control_targets"))}, + { + "primary_targets.battery.0", + "primary_targets.battery.1", + "control_targets.battery.0", + "control_targets.battery.1", + }, + ) + indexed_fields = [field for field in left.fields if field.name.startswith(("primary_targets", "control_targets"))] + self.assertTrue(all(field.provenance for field in indexed_fields)) + self.assertTrue(all(item in left.provenance for field in indexed_fields for item in field.provenance)) + self.assertFalse(left.materialization_readiness.ready) + self.assertEqual(left.materialization_readiness.blockers, ("config_projection_bindings_missing",)) + with self.assertRaises(AttributeError): + left.primary_targets[0].node_id = "changed" + + def test_provider_role_assignment_rejects_reference_and_negative_index(self): + """Only indexed primary/control assignments with non-negative indices exist.""" + with self.assertRaisesRegex(ValueError, "PRIMARY or CONTROL"): + ProviderRoleAssignment(AliasRole.REFERENCE, "battery", 0, "INV1") + with self.assertRaisesRegex(ValueError, "non-negative"): + ProviderRoleAssignment(AliasRole.PRIMARY, "battery", -1, "INV1") + + def test_indexed_role_indices_must_be_contiguous_per_group_and_role(self): + """A gap in one role sequence fails without affecting another sequence.""" + assignments = ( + ProviderRoleAssignment(AliasRole.PRIMARY, "battery", 0, "INV1"), + ProviderRoleAssignment(AliasRole.PRIMARY, "battery", 2, "INV1"), + ) + with self.assertRaisesRegex(AutoConfigCompileError, "indices must be contiguous"): + compile_auto_config((snapshot("gateway", role_assignments=assignments),)) + + def test_correlated_providers_may_share_one_index(self): + """Explicit strong identity correlation permits duplicate provider assertions.""" + gateway_identity = ProviderIdentityAlias("serial", "SER123", "gw") + cloud_identity = ProviderIdentityAlias("serial", "SER123", "cloud") + gateway_role = ProviderRoleAssignment(AliasRole.PRIMARY, "battery", 0, "gw") + cloud_role = ProviderRoleAssignment(AliasRole.PRIMARY, "battery", 0, "cloud") + plan = compile_auto_config( + ( + snapshot("gateway", node_id="gw", identity_aliases=(gateway_identity,), role_assignments=(gateway_role,)), + snapshot("cloud", node_id="cloud", identity_aliases=(cloud_identity,), role_assignments=(cloud_role,)), + ) + ) + + self.assertEqual(len(plan.primary_targets), 1) + self.assertEqual(plan.primary_targets[0].node_id, "identity:serial:SER123") + self.assertEqual({item.provider_id for item in plan.primary_targets[0].provenance}, {"gateway", "cloud"}) + + def test_uncorrelated_providers_conflicting_at_one_index_fail_closed(self): + """Equal role slots cannot select unrelated provider-local nodes.""" + gateway_role = ProviderRoleAssignment(AliasRole.PRIMARY, "battery", 0, "gw") + cloud_role = ProviderRoleAssignment(AliasRole.PRIMARY, "battery", 0, "cloud") + with self.assertRaisesRegex(AutoConfigCompileError, "conflicting primary target"): + compile_auto_config( + ( + snapshot("gateway", node_id="gw", role_assignments=(gateway_role,)), + snapshot("cloud", node_id="cloud", role_assignments=(cloud_role,)), + ) + ) + + def test_same_node_may_fill_multiple_indices(self): + """An EMS aggregate can fan one canonical node out over several indices.""" + assignments = ( + ProviderRoleAssignment(AliasRole.PRIMARY, "ems", 0, "EMS"), + ProviderRoleAssignment(AliasRole.PRIMARY, "ems", 1, "EMS"), + ) + plan = compile_auto_config((snapshot("ge-cloud", node_id="EMS", role_assignments=assignments),)) + + self.assertEqual([target.node_id for target in plan.primary_targets], ["provider:ge-cloud:EMS", "provider:ge-cloud:EMS"]) + + def test_legacy_and_indexed_role_assignments_cannot_mix(self): + """A plan must use exactly one target-addressing model.""" + legacy = ProviderAlias("battery", "INV1", frozenset((AliasRole.PRIMARY,))) + indexed = ProviderRoleAssignment(AliasRole.CONTROL, "battery", 0, "INV1") + with self.assertRaisesRegex(AutoConfigCompileError, "cannot be mixed"): + compile_auto_config((snapshot("gateway", aliases=(legacy,), role_assignments=(indexed,)),)) + + def test_multiple_legacy_assignments_are_ambiguous_even_when_correlated(self): + """The singular compatibility field represents exactly one assertion.""" + gateway_alias = ProviderAlias("battery", "gw", frozenset((AliasRole.PRIMARY,))) + cloud_alias = ProviderAlias("battery", "cloud", frozenset((AliasRole.PRIMARY,))) + gateway_identity = ProviderIdentityAlias("serial", "SER123", "gw") + cloud_identity = ProviderIdentityAlias("serial", "SER123", "cloud") + with self.assertRaisesRegex(AutoConfigCompileError, "ambiguous legacy primary"): + compile_auto_config( + ( + snapshot("gateway", node_id="gw", aliases=(gateway_alias,), identity_aliases=(gateway_identity,)), + snapshot("cloud", node_id="cloud", aliases=(cloud_alias,), identity_aliases=(cloud_identity,)), + ) + ) + + def test_reference_only_plan_is_explicitly_shadow_only(self): + """Reference discovery compiles but exposes every write blocker.""" + reference = ProviderAlias("battery", "INV1") + plan = compile_auto_config((snapshot("gateway", aliases=(reference,)),)) + + self.assertFalse(plan.materialization_readiness.ready) + self.assertEqual( + plan.materialization_readiness.blockers, + ( + "indexed_primary_targets_missing", + "indexed_control_targets_missing", + "config_projection_bindings_missing", + ), + ) + self.assertIsNone(plan.primary_target) + self.assertIsNone(plan.control_target) + class TestInvalidationStateMachine(unittest.TestCase): """Invalidations coalesce without losing freshness or last-known-good state.""" @@ -265,8 +472,9 @@ def reader(): self.assertEqual(run.attempts, 2) self.assertEqual(state["calls"], 2) self.assertEqual(dict(run.plan.provider_generations), {"gateway": 2}) - self.assertEqual([dict(request.plan.provider_generations) for request in requests], [{"gateway": 2}]) - self.assertEqual(run.materializations, 1) + self.assertEqual(requests, []) + self.assertEqual(run.materializations, 0) + self.assertFalse(run.plan.materialization_readiness.ready) self.assertFalse(run.pending) def test_all_accepted_invalidation_causes_survive_coalescing(self): @@ -395,7 +603,7 @@ def test_unavailable_active_provider_cannot_materialize_destructive_removal(self self.assertEqual(failed.status, CompileStatus.STALE) self.assertIs(failed.plan, last_known_good) - self.assertEqual(len(requests), 1) + self.assertEqual(len(requests), 0) self.assertEqual( set(dict(failed.plan.provider_generations)), {"gateway", "cloud"}, @@ -406,38 +614,24 @@ def test_unavailable_active_provider_cannot_materialize_destructive_removal(self ) self.assertTrue(failed.pending) - def test_materialization_failure_is_retryable_without_new_generation(self): - """Caller-driven retry can materialize the same complete generation.""" + def test_shadow_only_plan_never_invokes_materializer(self): + """A not-ready plan remains observable without reaching a write callback.""" reader = MutableReader(snapshot("gateway", generation=1)) requests = [] def materialize(request): - """Fail once, then accept the exact same semantic plan.""" + """Record any unsafe hand-off to make the test fail.""" requests.append(request) - if len(requests) == 1: - raise RuntimeError("temporary config store failure") compiler = LatticeAutoConfigCompiler({"gateway": reader}, materialize) - failed = compiler.drain() - - self.assertEqual(failed.status, CompileStatus.STALE) - self.assertIsNone(failed.plan) - self.assertEqual(failed.materializations, 0) - self.assertTrue(failed.pending) - self.assertIn( - "materialization_failed", - {issue.code for issue in failed.issues}, - ) - - recovered = compiler.drain() + run = compiler.drain() - self.assertEqual(recovered.status, CompileStatus.FRESH) - self.assertIsNotNone(recovered.plan) - self.assertEqual(recovered.materializations, 1) - self.assertFalse(recovered.pending) - self.assertEqual(reader.calls, 2) - self.assertEqual(len(requests), 2) - self.assertEqual(requests[0].plan.digest, requests[1].plan.digest) + self.assertEqual(run.status, CompileStatus.FRESH) + self.assertIsNotNone(run.plan) + self.assertFalse(run.plan.materialization_readiness.ready) + self.assertEqual(run.materializations, 0) + self.assertFalse(run.pending) + self.assertEqual(requests, []) def test_unchanged_digest_skips_materialization(self): """A newer generation with identical semantics updates provenance only.""" @@ -450,13 +644,14 @@ def test_unchanged_digest_skips_materialization(self): self.assertTrue(compiler.invalidate("gateway", 2, "heartbeat refresh")) second = compiler.drain() - self.assertEqual(first.materializations, 1) + self.assertEqual(first.materializations, 0) self.assertEqual(second.materializations, 0) - self.assertEqual(len(requests), 1) + self.assertEqual(len(requests), 0) + self.assertEqual(first.plan.digest, second.plan.digest) self.assertEqual(dict(second.plan.provider_generations), {"gateway": 2}) - def test_materializer_feedback_token_cannot_recompile(self): - """A materializer-caused integration event is not a feedback loop.""" + def test_shadow_plan_cannot_create_materializer_feedback(self): + """A blocked hand-off cannot cause an integration feedback loop.""" reader = MutableReader(snapshot("gateway", generation=1)) feedback_results = [] holder = {} @@ -469,11 +664,85 @@ def materialize(request): holder["compiler"] = compiler run = compiler.drain() - self.assertEqual(feedback_results, [False]) + self.assertEqual(feedback_results, []) self.assertEqual(run.attempts, 1) - self.assertEqual(run.materializations, 1) + self.assertEqual(run.materializations, 0) self.assertFalse(run.pending) + def test_ready_plan_retries_after_materializer_failure(self): + """A failed hand-off is not treated as a successful materialization.""" + requests = [] + + def materialize(request): + """Fail the first hand-off and accept the retry.""" + requests.append(request) + if len(requests) == 1: + raise RuntimeError("temporary write failure") + + compiler = LatticeAutoConfigCompiler(materializer=materialize) + plan = ready_plan() + + count, issues = compiler._materialize_if_changed(plan) + self.assertEqual(count, 0) + self.assertEqual([issue.code for issue in issues], ["materialization_failed"]) + + count, issues = compiler._materialize_if_changed(plan) + self.assertEqual(count, 1) + self.assertEqual(issues, ()) + self.assertEqual(len(requests), 2) + + def test_ready_plan_unchanged_digest_skips_materialization(self): + """Bookkeeping-only generation changes do not repeat a ready hand-off.""" + requests = [] + compiler = LatticeAutoConfigCompiler(materializer=requests.append) + plan = ready_plan() + + count, issues = compiler._materialize_if_changed(plan) + self.assertEqual((count, issues), (1, ())) + compiler._active_plan = plan + newer_generation = replace( + plan, + provider_generations=(("gateway", 2),), + ) + + count, issues = compiler._materialize_if_changed(newer_generation) + self.assertEqual((count, issues), (0, ())) + self.assertEqual(len(requests), 1) + + def test_ready_plan_suppresses_materializer_feedback_token(self): + """A ready hand-off cannot invalidate itself through its feedback token.""" + feedback_results = [] + holder = {} + + def materialize(request): + """Echo the hand-off token through the provider invalidation API.""" + feedback_results.append( + holder["compiler"].invalidate( + "gateway", + 2, + "materialized config observed", + request.feedback_token, + ) + ) + + compiler = LatticeAutoConfigCompiler( + {"gateway": MutableReader(snapshot("gateway"))}, + materialize, + ) + holder["compiler"] = compiler + + count, issues = compiler._materialize_if_changed(ready_plan()) + + self.assertEqual((count, issues), (1, ())) + self.assertEqual(feedback_results, [False]) + self.assertTrue( + compiler.invalidate( + "gateway", + 2, + "independent provider change", + ) + ) + def test_every_attempt_fresh_reads_all_providers(self): """Independent invalidations still re-read the complete provider set.""" gateway = MutableReader(snapshot("gateway", generation=1))