From c8db618915c5e9d41ed0f47a038b475af35fa872 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Mon, 27 Jul 2026 23:30:23 +0100 Subject: [PATCH] feat: project lattice fragments into auto config --- apps/predbat/lattice_autoconfig.py | 643 ++++++++++++++- apps/predbat/tests/test_lattice_autoconfig.py | 770 ++++++++++++++++++ 2 files changed, 1411 insertions(+), 2 deletions(-) diff --git a/apps/predbat/lattice_autoconfig.py b/apps/predbat/lattice_autoconfig.py index 847ff438f..42a24c957 100644 --- a/apps/predbat/lattice_autoconfig.py +++ b/apps/predbat/lattice_autoconfig.py @@ -42,6 +42,28 @@ class AliasRole(Enum): CONTROL = "control" +class ProjectionValueKind(Enum): + """Kind of one provider-published PredBat configuration value.""" + + ENTITY = "entity" + CONSTANT = "constant" + NONE = "none" + + +class ProjectionRouting(Enum): + """Relationship between projected values and selected indexed targets.""" + + LEAF = "leaf" + COORDINATOR = "coordinator" + + +class ProjectionCardinality(Enum): + """Shape of one projected PredBat configuration argument.""" + + SCALAR = "scalar" + PER_INDEX = "per_index" + + class CompileStatus(Enum): """Observable compiler state after a drain.""" @@ -121,6 +143,136 @@ def __post_init__(self): object.__setattr__(self, "node_id", self.node_id.strip()) +@dataclass(frozen=True) +class ProviderProjectionValue: + """One ordered provider-local entity, constant, or explicit None value.""" + + node_id: str + kind: ProjectionValueKind + value: object = None + capability: Optional[str] = None + identity_kind: Optional[str] = None + identity_value: Optional[str] = None + access_path_id: Optional[str] = None + + def __post_init__(self): + """Validate the value and normalize its optional source selectors.""" + if not isinstance(self.node_id, str) or not self.node_id.strip(): + raise ValueError("projection value node_id must be a non-empty string") + if not isinstance(self.kind, ProjectionValueKind): + raise ValueError("projection value kind must be a ProjectionValueKind") + if self.kind is ProjectionValueKind.ENTITY: + if not isinstance(self.value, str) or not self.value.strip(): + raise ValueError("projection entity value must be a non-empty string") + if not isinstance(self.capability, str) or not self.capability.strip(): + raise ValueError("projection entity value must name a capability") + value = self.value.strip() + elif self.kind is ProjectionValueKind.CONSTANT: + if not isinstance(self.value, (str, int, float, bool)) or self.value is None: + raise ValueError("projection constant must be a non-None JSON scalar") + value = self.value + _canonical_json(value) + else: + if self.value is not None: + raise ValueError("projection None value must carry value=None") + value = None + + capability = self.capability + if capability is not None: + if not isinstance(capability, str) or not capability.strip(): + raise ValueError("projection capability must be a non-empty string") + capability = capability.strip() + + identity_kind = self.identity_kind + identity_value = self.identity_value + if (identity_kind is None) != (identity_value is None): + raise ValueError("projection identity selector requires both kind and value") + if identity_kind is not None: + if not isinstance(identity_kind, str) or not identity_kind.strip(): + raise ValueError("projection identity kind must be a non-empty string") + if not isinstance(identity_value, str) or not identity_value.strip(): + raise ValueError("projection identity value must be a non-empty string") + identity_kind = identity_kind.strip().lower() + identity_value = identity_value.strip() + + access_path_id = self.access_path_id + if access_path_id is not None: + if not isinstance(access_path_id, str) or not access_path_id.strip(): + raise ValueError("projection access_path_id must be a non-empty string") + access_path_id = access_path_id.strip() + + object.__setattr__(self, "node_id", self.node_id.strip()) + object.__setattr__(self, "value", value) + object.__setattr__(self, "capability", capability) + object.__setattr__(self, "identity_kind", identity_kind) + object.__setattr__(self, "identity_value", identity_value) + object.__setattr__(self, "access_path_id", access_path_id) + + +@dataclass(frozen=True) +class ProviderConfigProjection: + """Generic provider contract for one PredBat configuration argument.""" + + argument: str + role: AliasRole + group: str + routing: ProjectionRouting + cardinality: ProjectionCardinality + values: tuple + required: bool = True + transforms: tuple = () + + def __post_init__(self): + """Validate and normalize one immutable projection declaration.""" + if not isinstance(self.argument, str) or not self.argument.strip(): + raise ValueError("projection argument must be a non-empty string") + if self.role not in (AliasRole.PRIMARY, AliasRole.CONTROL): + raise ValueError("projection role must be PRIMARY or CONTROL") + if not isinstance(self.group, str) or not self.group.strip(): + raise ValueError("projection group must be a non-empty string") + if not isinstance(self.routing, ProjectionRouting): + raise ValueError("projection routing must be a ProjectionRouting") + if not isinstance(self.cardinality, ProjectionCardinality): + raise ValueError("projection cardinality must be a ProjectionCardinality") + if not isinstance(self.required, bool): + raise ValueError("projection required must be a boolean") + values = tuple(self.values) + if not values or any(not isinstance(value, ProviderProjectionValue) for value in values): + raise ValueError("projection values must contain ProviderProjectionValue values") + if self.cardinality is ProjectionCardinality.SCALAR and len(values) != 1: + raise ValueError("scalar projection must contain exactly one value") + transforms = tuple(self.transforms) + if any(not isinstance(transform, str) or not transform.strip() for transform in transforms): + raise ValueError("projection transforms must be non-empty strings") + transforms = tuple(transform.strip() for transform in transforms) + if len(transforms) != len(set(transforms)): + raise ValueError("projection transforms must be unique") + object.__setattr__(self, "argument", self.argument.strip()) + object.__setattr__(self, "group", self.group.strip()) + object.__setattr__(self, "values", values) + object.__setattr__(self, "transforms", transforms) + + +@dataclass(frozen=True) +class UserConfigOverride: + """Explicit user-owned final value for one projected configuration argument.""" + + argument: str + value: object + source_path: str + + def __post_init__(self): + """Detach the override value and retain its explicit source path.""" + if not isinstance(self.argument, str) or not self.argument.strip(): + raise ValueError("override argument must be a non-empty string") + if not isinstance(self.source_path, str) or not self.source_path.strip(): + raise ValueError("override source_path must be a non-empty string") + value = _freeze(copy.deepcopy(self.value)) + object.__setattr__(self, "argument", self.argument.strip()) + object.__setattr__(self, "value", value) + object.__setattr__(self, "source_path", self.source_path.strip()) + + @dataclass(frozen=True) class ProviderSnapshot: """One integration's immutable generation of health, topology, and aliases.""" @@ -132,6 +284,7 @@ class ProviderSnapshot: aliases: tuple = () identity_aliases: tuple = () role_assignments: tuple = () + config_projections: tuple = () def __post_init__(self): """Validate scalar fields and detach caller-owned mutable data.""" @@ -152,11 +305,15 @@ def __post_init__(self): 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") + config_projections = tuple(self.config_projections) + if any(not isinstance(projection, ProviderConfigProjection) for projection in config_projections): + raise ValueError("config_projections must contain ProviderConfigProjection 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) + object.__setattr__(self, "config_projections", config_projections) @dataclass(frozen=True) @@ -253,6 +410,43 @@ class AutoConfigField: provenance: tuple +@dataclass(frozen=True) +class ProjectionCandidate: + """One normalized provider candidate for a projected argument.""" + + argument: str + provider_id: str + generation: int + role: str + group: str + routing: str + cardinality: str + required: bool + values: tuple + value_kinds: tuple + capabilities: tuple + identity_selectors: tuple + access_path_ids: tuple + transforms: tuple + specificities: tuple + provenance: tuple + + +@dataclass(frozen=True) +class ProjectedConfigArgument: + """Effective immutable PredBat argument plus all provider candidates.""" + + name: str + value: object + cardinality: str + required: bool + transforms: tuple + provenance: tuple + candidate_provenance: tuple + candidates: tuple + override_source: Optional[str] = None + + @dataclass(frozen=True) class AutoConfigPlan: """Deterministic immutable result of compiling all usable fragments.""" @@ -267,6 +461,8 @@ class AutoConfigPlan: primary_target: Optional[str] control_target: Optional[str] materialization_readiness: MaterializationReadiness + config_arguments: tuple + projected_config: Mapping fields: tuple provenance: tuple provider_generations: tuple @@ -350,6 +546,31 @@ def _semantic_topology(site): return topology +def _projection_payload(projection): + """Return one provider projection in deterministic plain form.""" + return { + "argument": projection.argument, + "role": projection.role.value, + "group": projection.group, + "routing": projection.routing.value, + "cardinality": projection.cardinality.value, + "required": projection.required, + "transforms": projection.transforms, + "values": [ + { + "node_id": value.node_id, + "kind": value.kind.value, + "value": value.value, + "capability": value.capability, + "identity_kind": value.identity_kind, + "identity_value": value.identity_value, + "access_path_id": value.access_path_id, + } + for value in projection.values + ], + } + + def _fingerprint_snapshot(snapshot): """Bind a provider generation to exactly one health/fragment/alias value.""" aliases = [ @@ -377,6 +598,19 @@ def _fingerprint_snapshot(snapshot): } for assignment in sorted(snapshot.role_assignments, key=lambda item: (item.group, item.role.value, item.index, item.node_id)) ] + config_projections = [ + _projection_payload(projection) + for projection in sorted( + snapshot.config_projections, + key=lambda item: ( + item.argument, + item.group, + item.role.value, + item.routing.value, + item.cardinality.value, + ), + ) + ] payload = { "provider": snapshot.provider_id, "generation": snapshot.generation, @@ -385,6 +619,7 @@ def _fingerprint_snapshot(snapshot): "aliases": aliases, "identity_aliases": identity_aliases, "role_assignments": role_assignments, + "config_projections": config_projections, } return hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest() @@ -682,6 +917,359 @@ def _compile_roles(snapshots, bindings, identity_bindings, provider_nodes, canon return role_bindings, primary_targets, control_targets, legacy_targets, readiness +def _projection_sort_key(projection): + """Return the stable ordering key for provider projection declarations.""" + return ( + projection.argument, + projection.group, + projection.role.value, + projection.routing.value, + projection.cardinality.value, + _canonical_json(_projection_payload(projection)), + ) + + +def _projection_topology_sources(document, provider_id): + """Index provider-owned access paths and capability-to-path bindings.""" + access_paths = {} + capabilities = set() + for node in document.get("nodes", ()): + node_id = str(node["id"]) + for access_path in node.get("accessPaths", ()): + path_id = str(access_path["id"]) + owner = access_path.get("provider", provider_id) + access_paths[(node_id, path_id)] = owner + for capability in node.get("capabilities", ()): + capabilities.add( + ( + node_id, + str(capability["capability"]), + str(capability["accessPath"]), + ) + ) + return access_paths, capabilities + + +def _projection_override_value(override, candidate): + """Validate and return an override in the provider candidate's shape.""" + value = override.value + if candidate.cardinality == ProjectionCardinality.PER_INDEX.value: + if not isinstance(value, tuple): + raise AutoConfigCompileError("override {} must be an ordered per-index sequence".format(override.argument)) + if len(value) != len(candidate.values): + raise AutoConfigCompileError( + "override {} cardinality mismatch: expected {}, got {}".format( + override.argument, + len(candidate.values), + len(value), + ) + ) + values = value + else: + if isinstance(value, (tuple, Mapping)): + raise AutoConfigCompileError("override {} must be a scalar value".format(override.argument)) + values = (value,) + for item in values: + if not isinstance(item, (str, int, float, bool)) and item is not None: + raise AutoConfigCompileError("override {} contains a non-scalar value".format(override.argument)) + _canonical_json(item) + if candidate.required and any(item is None for item in values): + raise AutoConfigCompileError("override {} leaves a required slot empty".format(override.argument)) + return value + + +def _compile_config_projections( + snapshots, + primary_targets, + control_targets, + provider_nodes, + canonical_by_node, + documents_by_provider, + user_overrides, +): + """Compile provider projection candidates into effective PredBat arguments.""" + targets_by_role_group = {} + for role, targets in ( + (AliasRole.PRIMARY, primary_targets), + (AliasRole.CONTROL, control_targets), + ): + for target in targets: + targets_by_role_group.setdefault((role, target.group), []).append(target) + targets_by_role_group = {key: tuple(sorted(targets, key=lambda item: item.index)) for key, targets in targets_by_role_group.items()} + + candidates_by_argument = {} + for snapshot in snapshots: + access_paths, capabilities = _projection_topology_sources( + documents_by_provider[snapshot.provider_id], + snapshot.provider_id, + ) + identity_assertions = {(alias.kind, alias.value, alias.node_id) for alias in snapshot.identity_aliases} + seen_arguments = set() + for projection_index, projection in enumerate(sorted(snapshot.config_projections, key=_projection_sort_key)): + if projection.argument in seen_arguments: + raise AutoConfigCompileError( + "provider {} repeats projection argument {}".format( + snapshot.provider_id, + projection.argument, + ) + ) + seen_arguments.add(projection.argument) + targets = targets_by_role_group.get( + (projection.role, projection.group), + (), + ) + if not targets: + raise AutoConfigCompileError( + "projection {} has no selected {} targets in group {}".format( + projection.argument, + projection.role.value, + projection.group, + ) + ) + if projection.cardinality is ProjectionCardinality.PER_INDEX and len(projection.values) != len(targets): + raise AutoConfigCompileError( + "projection {} cardinality mismatch: expected {}, got {}".format( + projection.argument, + len(targets), + len(projection.values), + ) + ) + if projection.routing is ProjectionRouting.COORDINATOR and len({target.node_id for target in targets}) != 1: + raise AutoConfigCompileError("coordinator projection {} requires one canonical target".format(projection.argument)) + + values = [] + kinds = [] + specificities = [] + provenance = [] + for value_index, value in enumerate(projection.values): + if value.node_id not in provider_nodes[snapshot.provider_id]: + raise AutoConfigCompileError( + "projection {} targets unknown provider-local node {}".format( + projection.argument, + value.node_id, + ) + ) + if value.kind is ProjectionValueKind.ENTITY: + matching_capabilities = { + access_path_id + for ( + node_id, + capability, + access_path_id, + ) in capabilities + if node_id == value.node_id and capability == value.capability + } + if not matching_capabilities: + raise AutoConfigCompileError( + "projection {} capability {} is not published by node {}".format( + projection.argument, + value.capability, + value.node_id, + ) + ) + if value.access_path_id is not None and value.access_path_id not in matching_capabilities: + raise AutoConfigCompileError( + "projection {} capability {} is not bound to access path {}".format( + projection.argument, + value.capability, + value.access_path_id, + ) + ) + target_index = value_index if projection.cardinality is ProjectionCardinality.PER_INDEX else 0 + canonical_node_id = canonical_by_node[(snapshot.provider_id, value.node_id)] + target = targets[target_index] + if canonical_node_id != target.node_id: + raise AutoConfigCompileError( + "projection {} slot {} is unrelated to selected canonical target {}".format( + projection.argument, + target_index, + target.node_id, + ) + ) + + specificity = 0 + if value.identity_kind is not None: + identity = ( + value.identity_kind, + value.identity_value, + value.node_id, + ) + if identity not in identity_assertions: + raise AutoConfigCompileError( + "projection {} identity selector is not asserted by provider {}".format( + projection.argument, + snapshot.provider_id, + ) + ) + specificity = 1 + if value.access_path_id is not None: + owner = access_paths.get((value.node_id, value.access_path_id)) + if owner != snapshot.provider_id: + raise AutoConfigCompileError( + "projection {} access path {} is not provider-owned".format( + projection.argument, + value.access_path_id, + ) + ) + specificity = 2 + if projection.required and value.kind is ProjectionValueKind.NONE: + raise AutoConfigCompileError( + "projection {} leaves required slot {} empty".format( + projection.argument, + target_index, + ) + ) + + field_path = "/projected_config/{}".format(projection.argument) + if projection.cardinality is ProjectionCardinality.PER_INDEX: + field_path += "/{}".format(target_index) + values.append(value.value) + kinds.append(value.kind.value) + specificities.append(specificity) + provenance.append( + ( + FieldProvenance( + field_path, + snapshot.provider_id, + snapshot.generation, + "/config_projections/{}/values/{}".format( + projection_index, + value_index, + ), + ), + ) + ) + + candidate = ProjectionCandidate( + argument=projection.argument, + provider_id=snapshot.provider_id, + generation=snapshot.generation, + role=projection.role.value, + group=projection.group, + routing=projection.routing.value, + cardinality=projection.cardinality.value, + required=projection.required, + values=tuple(values), + value_kinds=tuple(kinds), + capabilities=tuple(value.capability for value in projection.values), + identity_selectors=tuple( + ( + value.identity_kind, + value.identity_value, + ) + if value.identity_kind is not None + else None + for value in projection.values + ), + access_path_ids=tuple(value.access_path_id for value in projection.values), + transforms=projection.transforms, + specificities=tuple(specificities), + provenance=tuple(provenance), + ) + candidates_by_argument.setdefault( + projection.argument, + [], + ).append(candidate) + + overrides = {} + for override in tuple(user_overrides): + if not isinstance(override, UserConfigOverride): + raise ValueError("user_overrides must contain UserConfigOverride values") + if override.argument in overrides: + raise AutoConfigCompileError("duplicate user override for {}".format(override.argument)) + overrides[override.argument] = override + unknown_overrides = sorted(set(overrides) - set(candidates_by_argument)) + if unknown_overrides: + raise AutoConfigCompileError("user override has no provider candidate: {}".format(", ".join(unknown_overrides))) + + arguments = [] + projected_config = {} + for argument, unordered_candidates in sorted(candidates_by_argument.items()): + candidates = tuple( + sorted( + unordered_candidates, + key=lambda item: (item.provider_id, item.generation), + ) + ) + metadata = { + ( + candidate.role, + candidate.group, + candidate.routing, + candidate.cardinality, + candidate.required, + candidate.transforms, + len(candidate.values), + ) + for candidate in candidates + } + if len(metadata) != 1: + raise AutoConfigCompileError("conflicting projection contract for {}".format(argument)) + first = candidates[0] + candidate_provenance = tuple(source for candidate in candidates for slot_sources in candidate.provenance for source in slot_sources) + override = overrides.get(argument) + selected_values = [] + selected_provenance = [] + if override is None: + for index in range(len(first.values)): + highest_specificity = max(candidate.specificities[index] for candidate in candidates) + selected = tuple(candidate for candidate in candidates if candidate.specificities[index] == highest_specificity) + kinds = {candidate.value_kinds[index] for candidate in selected} + if len(kinds) != 1: + raise AutoConfigCompileError( + "projection {} type mismatch at slot {}".format( + argument, + index, + ) + ) + canonical_values = {_canonical_json(candidate.values[index]) for candidate in selected} + if len(canonical_values) != 1: + raise AutoConfigCompileError( + "ambiguous multi-provider projection {} at slot {}".format( + argument, + index, + ) + ) + selected_values.append(selected[0].values[index]) + selected_provenance.extend(source for candidate in selected for source in candidate.provenance[index]) + if first.cardinality == ProjectionCardinality.PER_INDEX.value: + effective_value = tuple(selected_values) + else: + effective_value = selected_values[0] + override_source = None + else: + effective_value = _projection_override_value(override, first) + override_source = override.source_path + selected_provenance = [ + FieldProvenance( + "/projected_config/{}".format(argument), + "user_override", + 0, + override.source_path, + ) + ] + + effective_value = _freeze(effective_value) + projected_config[argument] = effective_value + arguments.append( + ProjectedConfigArgument( + name=argument, + value=effective_value, + cardinality=first.cardinality, + required=first.required, + transforms=first.transforms, + provenance=tuple(selected_provenance), + candidate_provenance=candidate_provenance, + candidates=candidates, + override_source=override_source, + ) + ) + return ( + tuple(arguments), + _freeze(projected_config), + ) + + def _field_provenance( snapshots, bindings, @@ -745,7 +1333,7 @@ def _field_provenance( return tuple(sorted(provenance, key=lambda item: (item.field_path, item.provider_id, item.generation, item.source_path))) -def compile_auto_config(snapshots): +def compile_auto_config(snapshots, user_overrides=()): """Compile usable provider snapshots into one deterministic immutable plan.""" snapshots = tuple(sorted(snapshots, key=lambda item: item.provider_id)) if not snapshots: @@ -755,6 +1343,7 @@ def compile_auto_config(snapshots): raise AutoConfigCompileError("duplicate provider snapshots are not allowed") documents = [] + documents_by_provider = {} provider_nodes = {} for snapshot in snapshots: document = decode_topology(_plain(snapshot.topology_fragment)) @@ -765,6 +1354,7 @@ def compile_auto_config(snapshots): raise AutoConfigCompileError("provider {} supplied fragment owned by {}".format(snapshot.provider_id, document_provider)) nodes = _document_node_ids(document, snapshot.provider_id) provider_nodes[snapshot.provider_id] = nodes + documents_by_provider[snapshot.provider_id] = document documents.append(document) documents, canonical_by_node, identity_bindings = _correlate_identities(snapshots, documents, provider_nodes) @@ -800,6 +1390,22 @@ def compile_auto_config(snapshots): control_target = legacy_targets[AliasRole.CONTROL] topology_snapshot = merge_topologies(documents) + config_arguments, projected_config = _compile_config_projections( + snapshots, + primary_targets, + control_targets, + provider_nodes, + canonical_by_node, + documents_by_provider, + user_overrides, + ) + if config_arguments: + blockers = tuple(blocker for blocker in readiness.blockers if blocker != "config_projection_bindings_missing") + ("atomic_materializer_missing",) + readiness = MaterializationReadiness( + ready=False, + blockers=blockers, + ) + fields = [] for binding in bindings: source = FieldProvenance( @@ -835,6 +1441,14 @@ def compile_auto_config(snapshots): target.provenance, ) ) + for argument in config_arguments: + fields.append( + AutoConfigField( + "config.{}".format(argument.name), + argument.value, + argument.provenance, + ) + ) fields = tuple(sorted(fields, key=lambda item: item.name)) semantic = { @@ -876,10 +1490,21 @@ def compile_auto_config(snapshots): "ready": readiness.ready, "blockers": readiness.blockers, }, + "config_arguments": [ + { + "name": argument.name, + "value": argument.value, + "cardinality": argument.cardinality, + "required": argument.required, + "transforms": argument.transforms, + } + for argument in config_arguments + ], + "projected_config": projected_config, "fields": [{"name": field.name, "value": field.value} for field in fields], } digest = hashlib.sha256(_canonical_json(semantic).encode("utf-8")).hexdigest() - provenance = _field_provenance( + base_provenance = _field_provenance( snapshots, bindings, identity_bindings, @@ -890,6 +1515,18 @@ def compile_auto_config(snapshots): topology_snapshot, canonical_by_node, ) + projection_provenance = tuple(source for argument in config_arguments for source in (argument.candidate_provenance + argument.provenance)) + provenance = tuple( + sorted( + set(base_provenance + projection_provenance), + key=lambda item: ( + item.field_path, + item.provider_id, + item.generation, + item.source_path, + ), + ) + ) return AutoConfigPlan( digest=digest, topology=_freeze(topology_snapshot.site), @@ -901,6 +1538,8 @@ def compile_auto_config(snapshots): primary_target=primary_target, control_target=control_target, materialization_readiness=readiness, + config_arguments=config_arguments, + projected_config=projected_config, fields=fields, provenance=provenance, provider_generations=tuple((snapshot.provider_id, snapshot.generation) for snapshot in snapshots), diff --git a/apps/predbat/tests/test_lattice_autoconfig.py b/apps/predbat/tests/test_lattice_autoconfig.py index 285fb4010..edc4965fc 100644 --- a/apps/predbat/tests/test_lattice_autoconfig.py +++ b/apps/predbat/tests/test_lattice_autoconfig.py @@ -16,11 +16,17 @@ CompileStatus, LatticeAutoConfigCompiler, MaterializationReadiness, + ProjectionCardinality, + ProjectionRouting, + ProjectionValueKind, + ProviderConfigProjection, ProviderAlias, ProviderHealth, ProviderIdentityAlias, + ProviderProjectionValue, ProviderRoleAssignment, ProviderSnapshot, + UserConfigOverride, compile_auto_config, ) @@ -75,6 +81,118 @@ def snapshot( ) +def multi_fragment(provider, node_ids): + """Build a provider fragment containing ordered independent inverter nodes.""" + nodes = [] + for index, node_id in enumerate(node_ids): + access_path = "{}-{}-path".format(provider, node_id) + nodes.append( + { + "id": node_id, + "kind": "inverter", + "deviceType": "hybrid", + "accessPaths": [ + { + "id": access_path, + "provider": provider, + "preference": 10, + } + ], + "capabilities": [ + { + "capability": "battery.target_soc", + "accessPath": access_path, + "ref": index + 1, + "shape": "setpoint", + "control": {"protocol": "mqtt"}, + } + ], + } + ) + return { + "topologyVersion": "0.3.0", + "scope": "fragment", + "docVersion": 1, + "producer": { + "name": provider, + "provider": provider, + "authority": 10, + }, + "nodes": nodes, + } + + +def projection_snapshot( + provider, + node_ids, + role_assignments, + config_projections, + identity_aliases=(), + generation=1, +): + """Build a provider snapshot carrying indexed roles and config projections.""" + return ProviderSnapshot( + provider_id=provider, + generation=generation, + health=ProviderHealth.HEALTHY, + topology_fragment=multi_fragment(provider, node_ids), + identity_aliases=identity_aliases, + role_assignments=role_assignments, + config_projections=config_projections, + ) + + +def projection_value( + node_id, + kind, + value=None, + capability=None, + identity=None, + access_path_id=None, +): + """Build one compact provider projection value for tests.""" + identity_kind, identity_value = identity or (None, None) + if kind is ProjectionValueKind.ENTITY and capability is None: + capability = "battery.target_soc" + return ProviderProjectionValue( + node_id=node_id, + kind=kind, + value=value, + capability=capability, + identity_kind=identity_kind, + identity_value=identity_value, + access_path_id=access_path_id, + ) + + +def config_projection( + argument, + values, + role=AliasRole.PRIMARY, + group="inverters", + routing=ProjectionRouting.LEAF, + cardinality=ProjectionCardinality.PER_INDEX, + required=True, + transforms=(), +): + """Build one generic provider projection declaration for tests.""" + return ProviderConfigProjection( + argument=argument, + role=role, + group=group, + routing=routing, + cardinality=cardinality, + values=values, + required=required, + transforms=transforms, + ) + + +def indexed_roles(node_ids): + """Select each node as both an indexed primary and control target.""" + return tuple(ProviderRoleAssignment(role, "inverters", index, node_id) for index, node_id in enumerate(node_ids) for role in (AliasRole.PRIMARY, AliasRole.CONTROL)) + + def ready_plan(): """Copy a compiled shadow plan into a future-materializer test harness.""" plan = compile_auto_config((snapshot("gateway"),)) @@ -403,6 +521,658 @@ def test_reference_only_plan_is_explicitly_shadow_only(self): self.assertIsNone(plan.control_target) +class TestConfigProjectionCompilation(unittest.TestCase): + """Provider contracts project selected indexed capabilities into config.""" + + def test_gateway_multi_aio_projects_ordered_leaf_arrays(self): + """Two Gateway-like AIO nodes produce deterministic per-index arrays.""" + nodes = ("AIO1", "AIO2") + gateway = projection_snapshot( + "gateway", + nodes, + indexed_roles(nodes), + ( + config_projection( + "battery_power", + tuple( + projection_value( + node, + ProjectionValueKind.ENTITY, + "sensor.gateway_{}_battery_power".format(node.lower()), + ) + for node in nodes + ), + ), + config_projection( + "inverter_type", + tuple( + projection_value( + node, + ProjectionValueKind.CONSTANT, + "GEC", + ) + for node in nodes + ), + ), + config_projection( + "num_inverters", + ( + projection_value( + nodes[0], + ProjectionValueKind.CONSTANT, + 2, + ), + ), + cardinality=ProjectionCardinality.SCALAR, + ), + ), + ) + + plan = compile_auto_config((gateway,)) + + self.assertEqual( + dict(plan.projected_config), + { + "battery_power": ( + "sensor.gateway_aio1_battery_power", + "sensor.gateway_aio2_battery_power", + ), + "inverter_type": ("GEC", "GEC"), + "num_inverters": 2, + }, + ) + self.assertEqual( + [argument.name for argument in plan.config_arguments], + ["battery_power", "inverter_type", "num_inverters"], + ) + self.assertEqual( + {field.name for field in plan.fields if field.name.startswith("config.")}, + { + "config.battery_power", + "config.inverter_type", + "config.num_inverters", + }, + ) + self.assertEqual( + plan.materialization_readiness.blockers, + ("atomic_materializer_missing",), + ) + self.assertFalse(plan.materialization_readiness.ready) + requests = [] + run = LatticeAutoConfigCompiler( + {"gateway": MutableReader(gateway)}, + requests.append, + ).drain() + self.assertEqual(run.materializations, 0) + self.assertEqual(requests, []) + + def test_ge_ems_coordinator_fans_out_entities_and_zero_constants(self): + """One selected EMS coordinator can publish an ordered fan-out array.""" + role_assignments = ( + ProviderRoleAssignment( + AliasRole.PRIMARY, + "inverters", + 0, + "BAT1", + ), + ProviderRoleAssignment( + AliasRole.PRIMARY, + "inverters", + 1, + "BAT2", + ), + ProviderRoleAssignment( + AliasRole.CONTROL, + "inverters", + 0, + "EMS", + ), + ProviderRoleAssignment( + AliasRole.CONTROL, + "inverters", + 1, + "EMS", + ), + ) + ems = projection_snapshot( + "gecloud", + ("EMS", "BAT1", "BAT2"), + role_assignments, + ( + config_projection( + "battery_power", + ( + projection_value( + "EMS", + ProjectionValueKind.ENTITY, + "sensor.gecloud_ems_battery_power", + ), + projection_value( + "EMS", + ProjectionValueKind.CONSTANT, + 0, + ), + ), + role=AliasRole.CONTROL, + routing=ProjectionRouting.COORDINATOR, + ), + config_projection( + "charge_start_time", + ( + projection_value( + "EMS", + ProjectionValueKind.ENTITY, + "select.gecloud_ems_charge_start", + ), + projection_value( + "EMS", + ProjectionValueKind.ENTITY, + "select.gecloud_ems_charge_start", + ), + ), + role=AliasRole.CONTROL, + routing=ProjectionRouting.COORDINATOR, + ), + ), + ) + + plan = compile_auto_config((ems,)) + + self.assertEqual( + plan.projected_config["battery_power"], + ("sensor.gecloud_ems_battery_power", 0), + ) + self.assertEqual( + plan.projected_config["charge_start_time"], + ( + "select.gecloud_ems_charge_start", + "select.gecloud_ems_charge_start", + ), + ) + battery_argument = next(argument for argument in plan.config_arguments if argument.name == "battery_power") + self.assertEqual( + battery_argument.candidates[0].routing, + ProjectionRouting.COORDINATOR.value, + ) + + def test_fox_projection_carries_transform_and_constant_flags(self): + """Fox-like power entities retain transform metadata and invert flags.""" + fox = projection_snapshot( + "fox", + ("FOX1",), + indexed_roles(("FOX1",)), + ( + config_projection( + "grid_power", + ( + projection_value( + "FOX1", + ProjectionValueKind.ENTITY, + "sensor.fox_fox1_grid_power", + ), + ), + transforms=("invert", "watts"), + ), + config_projection( + "grid_power_invert", + ( + projection_value( + "FOX1", + ProjectionValueKind.CONSTANT, + True, + ), + ), + ), + config_projection( + "inverter_type", + ( + projection_value( + "FOX1", + ProjectionValueKind.CONSTANT, + "FoxCloud", + ), + ), + ), + ), + ) + + plan = compile_auto_config((fox,)) + grid_argument = next(argument for argument in plan.config_arguments if argument.name == "grid_power") + + self.assertEqual(grid_argument.transforms, ("invert", "watts")) + self.assertEqual(plan.projected_config["grid_power_invert"], (True,)) + self.assertEqual(plan.projected_config["inverter_type"], ("FoxCloud",)) + + def test_solis_optional_projection_preserves_none_and_absence(self): + """Solis-like optional args distinguish explicit None from no binding.""" + solis = projection_snapshot( + "solis", + ("SOLIS1",), + indexed_roles(("SOLIS1",)), + ( + config_projection( + "givtcp_rest", + ( + projection_value( + "SOLIS1", + ProjectionValueKind.NONE, + ), + ), + cardinality=ProjectionCardinality.SCALAR, + required=False, + ), + config_projection( + "pause_mode", + ( + projection_value( + "SOLIS1", + ProjectionValueKind.NONE, + ), + ), + required=False, + ), + config_projection( + "inverter_type", + ( + projection_value( + "SOLIS1", + ProjectionValueKind.CONSTANT, + "SolisCloud", + ), + ), + ), + ), + ) + + plan = compile_auto_config((solis,)) + + self.assertIsNone(plan.projected_config["givtcp_rest"]) + self.assertEqual(plan.projected_config["pause_mode"], (None,)) + self.assertNotIn("idle_start_time", plan.projected_config) + + def test_cross_provider_identity_and_access_path_select_one_value(self): + """An explicit correlated access path beats a generic provider value.""" + gateway = projection_snapshot( + "gateway", + ("GW1",), + indexed_roles(("GW1",)), + ( + config_projection( + "battery_power", + ( + projection_value( + "GW1", + ProjectionValueKind.ENTITY, + "sensor.gateway_battery_power", + ), + ), + ), + ), + identity_aliases=(ProviderIdentityAlias("serial", "SER123", "GW1"),), + ) + cloud = projection_snapshot( + "cloud", + ("CLOUD1",), + (), + ( + config_projection( + "battery_power", + ( + projection_value( + "CLOUD1", + ProjectionValueKind.ENTITY, + "sensor.cloud_battery_power", + identity=("serial", "SER123"), + access_path_id="cloud-CLOUD1-path", + ), + ), + ), + ), + identity_aliases=( + ProviderIdentityAlias( + "serial", + "SER123", + "CLOUD1", + ), + ), + ) + + plan = compile_auto_config((gateway, cloud)) + argument = plan.config_arguments[0] + + self.assertEqual( + plan.projected_config["battery_power"], + ("sensor.cloud_battery_power",), + ) + self.assertEqual( + {candidate.provider_id for candidate in argument.candidates}, + {"gateway", "cloud"}, + ) + cloud_candidate = next(candidate for candidate in argument.candidates if candidate.provider_id == "cloud") + self.assertEqual( + cloud_candidate.capabilities, + ("battery.target_soc",), + ) + self.assertEqual( + cloud_candidate.identity_selectors, + (("serial", "SER123"),), + ) + self.assertEqual( + cloud_candidate.access_path_ids, + ("cloud-CLOUD1-path",), + ) + self.assertEqual( + {source.provider_id for source in argument.candidate_provenance}, + {"gateway", "cloud"}, + ) + + def test_user_override_wins_and_retains_provider_candidates(self): + """An explicit override resolves values without erasing candidates.""" + identity = "SER-OVERRIDE" + gateway = projection_snapshot( + "gateway", + ("GW1",), + indexed_roles(("GW1",)), + ( + config_projection( + "battery_power", + ( + projection_value( + "GW1", + ProjectionValueKind.ENTITY, + "sensor.gateway_candidate", + ), + ), + ), + ), + identity_aliases=(ProviderIdentityAlias("serial", identity, "GW1"),), + ) + cloud = projection_snapshot( + "cloud", + ("CLOUD1",), + (), + ( + config_projection( + "battery_power", + ( + projection_value( + "CLOUD1", + ProjectionValueKind.ENTITY, + "sensor.cloud_candidate", + ), + ), + ), + ), + identity_aliases=(ProviderIdentityAlias("serial", identity, "CLOUD1"),), + ) + override = UserConfigOverride( + "battery_power", + ["sensor.user_selected"], + "/apps.yaml/battery_power", + ) + + plan = compile_auto_config( + (gateway, cloud), + user_overrides=(override,), + ) + argument = plan.config_arguments[0] + + self.assertEqual( + plan.projected_config["battery_power"], + ("sensor.user_selected",), + ) + self.assertEqual(argument.override_source, "/apps.yaml/battery_power") + self.assertEqual( + {source.provider_id for source in argument.candidate_provenance}, + {"gateway", "cloud"}, + ) + self.assertEqual( + [source.provider_id for source in argument.provenance], + ["user_override"], + ) + + def test_projection_order_and_outputs_are_deterministic_and_immutable(self): + """Provider and declaration order cannot alter or mutate the plan.""" + projections = ( + config_projection( + "inverter_type", + ( + projection_value( + "INV1", + ProjectionValueKind.CONSTANT, + "GEC", + ), + ), + ), + config_projection( + "battery_power", + ( + projection_value( + "INV1", + ProjectionValueKind.ENTITY, + "sensor.inv1_battery_power", + ), + ), + ), + ) + left_snapshot = projection_snapshot( + "gateway", + ("INV1",), + indexed_roles(("INV1",)), + projections, + ) + right_snapshot = projection_snapshot( + "gateway", + ("INV1",), + indexed_roles(("INV1",)), + tuple(reversed(projections)), + ) + + left = compile_auto_config((left_snapshot,)) + right = compile_auto_config((right_snapshot,)) + + self.assertEqual(left.digest, right.digest) + self.assertEqual( + [argument.name for argument in left.config_arguments], + ["battery_power", "inverter_type"], + ) + with self.assertRaises(TypeError): + left.projected_config["battery_power"] = ("changed",) + with self.assertRaises(TypeError): + left.projected_config["battery_power"][0] = "changed" + + def test_projection_invalidation_recompiles_a_new_provider_generation(self): + """Any provider can invalidate and replace its projection generation.""" + + def projected(generation, entity_id): + """Build one generation of the provider's projected entity.""" + return projection_snapshot( + "gateway", + ("INV1",), + indexed_roles(("INV1",)), + ( + config_projection( + "battery_power", + ( + projection_value( + "INV1", + ProjectionValueKind.ENTITY, + entity_id, + ), + ), + ), + ), + generation=generation, + ) + + reader = MutableReader(projected(1, "sensor.gateway_battery_power_v1")) + compiler = LatticeAutoConfigCompiler({"gateway": reader}) + first = compiler.drain() + + reader.value = projected(2, "sensor.gateway_battery_power_v2") + self.assertTrue( + compiler.invalidate( + "gateway", + 2, + "projection binding changed", + ) + ) + second = compiler.drain() + + self.assertNotEqual(first.plan.digest, second.plan.digest) + self.assertEqual( + second.plan.projected_config["battery_power"], + ("sensor.gateway_battery_power_v2",), + ) + self.assertEqual( + dict(second.plan.provider_generations), + {"gateway": 2}, + ) + + def test_projection_gaps_types_conflicts_and_unrelated_nodes_fail_closed(self): + """Unsafe required gaps, shapes, sources, and assertions are rejected.""" + roles = indexed_roles(("INV1", "INV2")) + required_gap = projection_snapshot( + "gateway", + ("INV1", "INV2"), + roles, + ( + config_projection( + "battery_power", + ( + projection_value( + "INV1", + ProjectionValueKind.ENTITY, + "sensor.inv1", + ), + projection_value( + "INV2", + ProjectionValueKind.NONE, + ), + ), + ), + ), + ) + with self.assertRaisesRegex( + AutoConfigCompileError, + "required slot", + ): + compile_auto_config((required_gap,)) + + wrong_cardinality = projection_snapshot( + "gateway", + ("INV1", "INV2"), + roles, + ( + config_projection( + "battery_power", + ( + projection_value( + "INV1", + ProjectionValueKind.ENTITY, + "sensor.inv1", + ), + ), + ), + ), + ) + with self.assertRaisesRegex( + AutoConfigCompileError, + "cardinality mismatch", + ): + compile_auto_config((wrong_cardinality,)) + + unrelated = projection_snapshot( + "gateway", + ("INV1", "OTHER"), + indexed_roles(("INV1",)), + ( + config_projection( + "battery_power", + ( + projection_value( + "OTHER", + ProjectionValueKind.ENTITY, + "sensor.other", + ), + ), + ), + ), + ) + with self.assertRaisesRegex( + AutoConfigCompileError, + "unrelated", + ): + compile_auto_config((unrelated,)) + + identity = "SER-CONFLICT" + gateway = projection_snapshot( + "gateway", + ("GW1",), + indexed_roles(("GW1",)), + ( + config_projection( + "battery_power", + ( + projection_value( + "GW1", + ProjectionValueKind.ENTITY, + "sensor.gateway", + ), + ), + ), + ), + identity_aliases=(ProviderIdentityAlias("serial", identity, "GW1"),), + ) + cloud = projection_snapshot( + "cloud", + ("CLOUD1",), + (), + ( + config_projection( + "battery_power", + ( + projection_value( + "CLOUD1", + ProjectionValueKind.ENTITY, + "sensor.cloud", + ), + ), + ), + ), + identity_aliases=(ProviderIdentityAlias("serial", identity, "CLOUD1"),), + ) + with self.assertRaisesRegex( + AutoConfigCompileError, + "ambiguous multi-provider", + ): + compile_auto_config((gateway, cloud)) + + cloud_constant = projection_snapshot( + "cloud", + ("CLOUD1",), + (), + ( + config_projection( + "battery_power", + ( + projection_value( + "CLOUD1", + ProjectionValueKind.CONSTANT, + 0, + ), + ), + ), + ), + identity_aliases=(ProviderIdentityAlias("serial", identity, "CLOUD1"),), + ) + with self.assertRaisesRegex( + AutoConfigCompileError, + "type mismatch", + ): + compile_auto_config((gateway, cloud_constant)) + + class TestInvalidationStateMachine(unittest.TestCase): """Invalidations coalesce without losing freshness or last-known-good state."""