"""Strict P1B-3 policy loader for the 15-step kernel build input graph.""" from __future__ import annotations import ast import hashlib import json import re from dataclasses import dataclass from pathlib import Path from types import MappingProxyType from typing import Any, Mapping, Sequence import yaml from kernel.etl.build_input_contract import ( REFERENCE_SELECTION, ProductInputContractError, validate_product_input_source, ) from spec_runtime.metric_spec import load_metric API_VERSION = "specs.gwptd.io/v1alpha1" KIND = "KernelBuildInputManifest" PRODUCT_ONLY_KIND = "KernelProductInputManifest" EXPECTED_STEP_COUNT = 15 EXPECTED_ACTIVE_ENDPOINT_COUNT = 35 _DIGEST_SCHEMA_VERSION = 1 def _source_policy(counted: bool, eligible: bool, completeness: str) -> dict[str, object]: return {"countedActiveEndpoint": counted, "buildEligible": eligible, "completeness": completeness} _SOURCE_MODE_POLICIES = { "active_product_spec": _source_policy(True, True, "successful_complete_collection_run"), "excluded_product_spec": _source_policy(False, False, "excluded_from_kernel_build"), "external_reference": _source_policy(False, True, "producer_guarded_reference_snapshot"), "kernel_static_reference": _source_policy( False, True, "declared_relation_only_no_content_digest" ), "prepared_external": _source_policy(False, True, "validated_before_business_dml"), "legacy_recovery": _source_policy( False, True, "recovered_historical_payload_not_live_collection" ), "acknowledged_loss_ledger": _source_policy( False, True, "named_finite_allowlist_with_file_digest" ), "formula_contract": _source_policy(False, True, "file_digest_and_schema_gate"), "metric_spec": _source_policy(False, True, "file_digest_and_golden_gate"), } # Historical recovery is intentionally a bounded exceptional lane. Do not # make this a generic endpoint pattern: every admitted route must name its # intake relation, producer and parser lineage exactly, so a saved payload # cannot be relabelled as an arbitrary live collection input. # # marketplaceIds принадлежит контракту, а не резолверу. Раньше допустимые # маркетплейсы были зашиты в resolve_build_inputs как {6, 8} — набор Яндекса, # под который эту полосу писали первой. Из-за этого восстановление WB (mp 1) # отбивалось с причиной "recovery_raw_lineage_mismatch", хотя происхождение у # него сходилось полностью: сообщение указывало не на ту беду. _LEGACY_RECOVERY_CONTRACTS = { "wb.returns.legacy_recovery": { "id": "recovery:wb.returns.legacy_recovery", "tables": ["wb_returns"], "producer": "marketplace-collector-v3/scripts/recover_wb_returns_from_legacy.py", "collectorClass": "LegacyWbReturnsRecovery", "parserVersion": "legacy-recovery-2026-08-04", "sourceBusinessKeyPrefix": "legacy:lamoda_reports.mp_returns:", "sourceBusinessKeyAction": "return", "marketplaceIds": [1, 2], }, "ym.returns.legacy_recovery": { "id": "recovery:ym.returns.legacy_recovery", "tables": ["ym_returns"], "producer": "marketplace-collector-v3/scripts/recover_ym_returns_from_legacy.py", "collectorClass": "LegacyYmReturnsRecovery", "parserVersion": "legacy-recovery-2026-08-03", "sourceBusinessKeyPrefix": "legacy:lamoda_reports.mp_returns:", "sourceBusinessKeyAction": "return", "marketplaceIds": [6, 8], }, "ym.stats_orders.legacy_recovery": { "id": "recovery:ym.stats_orders.legacy_recovery", "tables": ["ym_stats_orders"], "producer": "marketplace-collector-v3/scripts/recover_ym_orders_from_legacy.py", "collectorClass": "LegacyYmOrdersRecovery", "parserVersion": "legacy-recovery-2026-08-04", "sourceBusinessKeyPrefix": "legacy:lamoda_reports.mp_orders_daily:", "sourceBusinessKeyAction": "order", "marketplaceIds": [6, 8], }, } _ACKNOWLEDGED_LOSS_LEDGER_CONTRACTS = { "ym.returns.acknowledged_missing_orders.v1": { "id": "ledger:ym.returns.acknowledged_missing_orders", "path": "specs/kernel/ym-acknowledged-missing-returns.json", "sourceEndpoint": "ym.returns.legacy_recovery", "expectedEntryCount": 79, "sha256": "484cda6ab4ee16dd8cb38845f18ced6907c425f92c98a683d62483b9a4f9491e", }, } ACKNOWLEDGED_YM_LOSS_LEDGER_ID = "ym.returns.acknowledged_missing_orders.v1" def _mutation_policy(writes: bool, destructive: bool, strategy: str) -> dict[str, object]: return {"writes": writes, "destructive": destructive, "strategy": strategy} _MUTATION_MODE_POLICIES = { "upsert": _mutation_policy(True, False, "insert_or_update"), "upsert_and_quarantine_reconcile": _mutation_policy(True, False, "identity_upsert_and_status_update"), "snapshot_reconcile": _mutation_policy(True, True, "authoritative_scope_replace_or_retract"), "scope_replace_and_upsert": _mutation_policy(True, True, "delete_bounded_scope_then_upsert"), "dedupe_delete": _mutation_policy(True, True, "deterministic_duplicate_retraction"), "derived_snapshot_replace": _mutation_policy(True, True, "replace_calculation_scope"), "validation_only": _mutation_policy(False, False, "fail_closed_read_gate"), } _ID_RE = re.compile(r"^[a-z][a-z0-9_.-]*:[a-z0-9][a-z0-9_.-]*$") _TABLE_RE = re.compile(r"^[a-z][a-z0-9_]*$") _RELATION_RE = re.compile(r"^[a-z][a-z0-9_]*\.[a-z][a-z0-9_]*$") _OUTPUT_RE = re.compile( r"^(?:gwptd_kernel\.[a-z][a-z0-9_]*|validation:[a-z][a-z0-9_]*)$" ) _ADAPTER_RE = re.compile( r"^[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)*:[a-z_][a-z0-9_]*$" ) class BuildInputManifestError(ValueError): """The input policy is malformed or drifted from executable code.""" def acknowledged_loss_ledger_source_endpoint(ledger_id: str) -> str: """Назвать endpoint, из которого ledger вообще может материализоваться. Операционным проверкам он нужен, чтобы отличить «источника ещё нет» от «источник есть, а запись пропала». Берём его из reviewed-контракта, а не повторяем строкой на стороне проверки. """ contract = _ACKNOWLEDGED_LOSS_LEDGER_CONTRACTS.get(ledger_id) if contract is None: raise BuildInputManifestError( f"ledgerId {ledger_id} is not a reviewed acknowledged-loss ledger" ) return str(contract["sourceEndpoint"]) class _UniqueKeyLoader(yaml.SafeLoader): """Safe loader which rejects duplicate mapping keys.""" def _construct_unique_mapping( loader: _UniqueKeyLoader, node: yaml.nodes.MappingNode, deep: bool = False, ) -> dict[Any, Any]: loader.flatten_mapping(node) result: dict[Any, Any] = {} for key_node, value_node in node.value: key = loader.construct_object(key_node, deep=deep) try: if key in result: raise BuildInputManifestError(f"duplicate YAML key: {key!r}") except TypeError as exc: raise BuildInputManifestError("YAML mapping key is not hashable") from exc result[key] = loader.construct_object(value_node, deep=deep) return result _UniqueKeyLoader.add_constructor( yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, _construct_unique_mapping, ) @dataclass(frozen=True) class BuildInputManifest: path: Path document: Mapping[str, Any] step_order: tuple[str, ...] active_endpoints: tuple[str, ...] policy_digest: str source_digests: Mapping[str, str] product_bindings: Mapping[str, Mapping[str, Any]] def _repo_root() -> Path: return Path(__file__).resolve().parents[3] def _default_manifest_path() -> Path: return _repo_root() / "specs" / "kernel" / "build-all-inputs.yaml" def _default_product_specs_path(root: Path) -> Path: return root / "specs" / "products" def _load_yaml(path: Path) -> Any: try: return yaml.load(path.read_text(encoding="utf-8"), Loader=_UniqueKeyLoader) except BuildInputManifestError: raise except (OSError, yaml.YAMLError) as exc: raise BuildInputManifestError(f"{path}: cannot load YAML: {exc}") from exc def _mapping(value: Any, label: str) -> dict[str, Any]: if not isinstance(value, dict): raise BuildInputManifestError(f"{label} must be a mapping") if not all(isinstance(key, str) for key in value): raise BuildInputManifestError(f"{label} keys must be strings") return value def _exact_keys(value: dict[str, Any], expected: set[str], label: str) -> None: actual = set(value) if actual != expected: raise BuildInputManifestError( f"{label} keys mismatch: missing={sorted(expected - actual)}, " f"extra={sorted(actual - expected)}" ) def _strings(value: Any, label: str, *, allow_empty: bool = False) -> list[str]: if not isinstance(value, list): raise BuildInputManifestError(f"{label} must be a list") if not allow_empty and not value: raise BuildInputManifestError(f"{label} must not be empty") if not all(isinstance(item, str) and item for item in value): raise BuildInputManifestError(f"{label} must contain non-empty strings") if len(value) != len(set(value)): raise BuildInputManifestError(f"{label} contains duplicates") return list(value) def _safe_relative_file(root: Path, value: Any, label: str) -> Path: text = str(value or "") relative = Path(text) if not text or relative.is_absolute() or ".." in relative.parts: raise BuildInputManifestError(f"{label} must be a safe repository-relative path") path = (root / relative).resolve() try: path.relative_to(root.resolve()) except ValueError as exc: raise BuildInputManifestError(f"{label} escapes repository root") from exc if not path.is_file(): raise BuildInputManifestError(f"{label} does not exist: {text}") return path def _file_digest(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _deep_freeze(value: Any) -> Any: """Seal the exact policy snapshot whose digest was computed.""" if isinstance(value, dict): return MappingProxyType({key: _deep_freeze(item) for key, item in value.items()}) if isinstance(value, list): return tuple(_deep_freeze(item) for item in value) return value def _validate_policies(document: dict[str, Any]) -> None: source_policies = _mapping(document["sourceModePolicies"], "sourceModePolicies") mutation_policies = _mapping(document["mutationModePolicies"], "mutationModePolicies") if source_policies != _SOURCE_MODE_POLICIES: raise BuildInputManifestError("sourceModePolicies drift from the bounded contract") if mutation_policies != _MUTATION_MODE_POLICIES: raise BuildInputManifestError("mutationModePolicies drift from the bounded contract") def _validate_sources( document: dict[str, Any], *, root: Path, products_path: Path, ) -> tuple[dict[str, dict[str, Any]], dict[str, str], dict[str, dict[str, Any]]]: raw_sources = document["sources"] if not isinstance(raw_sources, list) or not raw_sources: raise BuildInputManifestError("sources must be a non-empty list") sources: dict[str, dict[str, Any]] = {} source_digests: dict[str, str] = {} product_bindings: dict[str, dict[str, Any]] = {} product_endpoints: set[str] = set() product_modes = {"active_product_spec", "excluded_product_spec"} for index, raw_source in enumerate(raw_sources): source = _mapping(raw_source, f"sources[{index}]") source_id = str(source.get("id") or "") mode = str(source.get("mode") or "") if not _ID_RE.fullmatch(source_id): raise BuildInputManifestError(f"sources[{index}].id is invalid: {source_id!r}") if source_id in sources: raise BuildInputManifestError(f"duplicate source id: {source_id}") if mode not in _SOURCE_MODE_POLICIES: raise BuildInputManifestError(f"source {source_id} has unknown mode {mode!r}") if mode in product_modes: raw = _mapping( _load_yaml(products_path / f"{source.get('endpoint')}.yaml"), f"ProductSpec {source.get('endpoint')}", ) try: binding = validate_product_input_source( source, mode=mode, products_path=products_path, raw=raw ) except ProductInputContractError as exc: raise BuildInputManifestError(str(exc)) from exc endpoint = binding["endpoint"] if endpoint in product_endpoints: raise BuildInputManifestError(f"duplicate ProductSpec source: {endpoint}") product_endpoints.add(endpoint) product_bindings[endpoint] = binding source_digests[source_id] = binding["digest"] elif mode == "external_reference": keys = { "id", "mode", "inputMode", "selection", "table", "producer", "completenessGuard", } _exact_keys(source, keys, f"source {source_id}") if ( source.get("inputMode") != "reference" or source.get("selection") != REFERENCE_SELECTION ): raise BuildInputManifestError(f"source {source_id} reference semantics are invalid") if not _RELATION_RE.fullmatch(str(source.get("table") or "")): raise BuildInputManifestError(f"source {source_id}.table is invalid") producer = _safe_relative_file(root, source["producer"], f"{source_id}.producer") guard = source.get("completenessGuard") if not isinstance(guard, str) or not guard: raise BuildInputManifestError(f"source {source_id}.completenessGuard must be non-empty") source_digests[source_id] = _file_digest(producer) elif mode == "prepared_external": keys = {"id", "mode", "providerRef", "completenessGuard"} _exact_keys(source, keys, f"source {source_id}") if not _ADAPTER_RE.fullmatch(str(source.get("providerRef") or "")): raise BuildInputManifestError(f"source {source_id}.providerRef is invalid") guard = source.get("completenessGuard") if not isinstance(guard, str) or not guard: raise BuildInputManifestError(f"source {source_id}.completenessGuard must be non-empty") elif mode == "legacy_recovery": keys = { "id", "mode", "endpoint", "tables", "producer", "collectorClass", "parserVersion", "sourceBusinessKeyPrefix", "sourceBusinessKeyAction", "marketplaceIds", "completenessGuard", } _exact_keys(source, keys, f"source {source_id}") endpoint = str(source.get("endpoint") or "") contract = _LEGACY_RECOVERY_CONTRACTS.get(endpoint) if contract is None: raise BuildInputManifestError( f"source {source_id}.endpoint is not a reviewed legacy recovery route" ) for field in ("id", "tables", "producer", "collectorClass", "parserVersion"): if source.get(field) != contract[field]: raise BuildInputManifestError( f"source {source_id}.{field} does not match the reviewed {endpoint} contract" ) if source.get("sourceBusinessKeyPrefix") != contract["sourceBusinessKeyPrefix"]: raise BuildInputManifestError( f"source {source_id}.sourceBusinessKeyPrefix is invalid" ) if source.get("sourceBusinessKeyAction") != contract["sourceBusinessKeyAction"]: raise BuildInputManifestError( f"source {source_id}.sourceBusinessKeyAction is invalid" ) marketplace_ids = source.get("marketplaceIds") if ( not isinstance(marketplace_ids, (list, tuple)) or [int(mp) for mp in marketplace_ids] != list(contract["marketplaceIds"]) ): raise BuildInputManifestError( f"source {source_id}.marketplaceIds does not match the reviewed " f"{endpoint} contract" ) if source.get("completenessGuard") != ( "recovered_historical_payload_not_live_collection" ): raise BuildInputManifestError( f"source {source_id}.completenessGuard is invalid" ) producer = _safe_relative_file(root, contract["producer"], f"{source_id}.producer") source_digests[source_id] = _file_digest(producer) elif mode == "acknowledged_loss_ledger": keys = { "id", "mode", "ledgerId", "path", "sourceEndpoint", "expectedEntryCount", "sha256", "completenessGuard", } _exact_keys(source, keys, f"source {source_id}") ledger_id = str(source.get("ledgerId") or "") contract = _ACKNOWLEDGED_LOSS_LEDGER_CONTRACTS.get(ledger_id) if contract is None: raise BuildInputManifestError( f"source {source_id}.ledgerId is not a reviewed acknowledged-loss ledger" ) for field in ("id", "path", "sourceEndpoint", "expectedEntryCount", "sha256"): if source.get(field) != contract[field]: raise BuildInputManifestError( f"source {source_id}.{field} does not match the reviewed {ledger_id} contract" ) if source.get("completenessGuard") != "named_finite_allowlist_with_file_digest": raise BuildInputManifestError( f"source {source_id}.completenessGuard is invalid" ) path = _safe_relative_file(root, contract["path"], f"{source_id}.path") if _file_digest(path) != contract["sha256"]: raise BuildInputManifestError( f"source {source_id} ledger file differs from its reviewed sha256" ) ledger = _mapping(_load_yaml(path), f"acknowledged-loss ledger {path}") if ledger.get("ledgerId") != ledger_id: raise BuildInputManifestError(f"source {source_id} ledgerId differs from its file") entries = ledger.get("entries") if ( not isinstance(entries, list) or ledger.get("expectedEntryCount") != contract["expectedEntryCount"] or len(entries) != contract["expectedEntryCount"] ): raise BuildInputManifestError( f"source {source_id} ledger is not the reviewed finite entry set" ) source_digests[source_id] = _file_digest(path) elif mode == "kernel_static_reference": keys = {"id", "mode", "table", "consistency", "completenessGuard"} _exact_keys(source, keys, f"source {source_id}") relation = str(source.get("table") or "") if not _RELATION_RE.fullmatch(relation) or not relation.startswith( "gwptd_kernel." ): raise BuildInputManifestError( f"source {source_id}.table must be a gwptd_kernel relation" ) # This source mode deliberately makes no content-completeness claim. # P1B-5 must later bind these reads to a generation/snapshot token. if source.get("consistency") != "caller_transaction_visibility": raise BuildInputManifestError( f"source {source_id}.consistency is invalid" ) if source.get("completenessGuard") != ( "declared_relation_only_no_content_digest" ): raise BuildInputManifestError( f"source {source_id}.completenessGuard is invalid" ) elif mode == "formula_contract": _exact_keys(source, {"id", "mode", "path"}, f"source {source_id}") path = _safe_relative_file(root, source["path"], f"{source_id}.path") formula = _mapping(_load_yaml(path), f"formula contract {path}") required = {"constants", "cost_price", "rating", "turnover", "invariants"} if not required.issubset(formula): missing = sorted(required - set(formula)) raise BuildInputManifestError(f"formula contract missing sections: {missing}") source_digests[source_id] = _file_digest(path) else: _exact_keys(source, {"id", "mode", "metric", "path"}, f"source {source_id}") metric = str(source.get("metric") or "") path = _safe_relative_file(root, source["path"], f"{source_id}.path") if path.name != f"{metric}.yaml": raise BuildInputManifestError(f"source {source_id} MetricSpec path/name mismatch") try: load_metric(metric, path.parent) except Exception as exc: raise BuildInputManifestError(f"source {source_id} MetricSpec is invalid: {exc}") from exc source_digests[source_id] = _file_digest(path) sources[source_id] = source active_count = sum(source["mode"] == "active_product_spec" for source in sources.values()) declared_count = document["metadata"]["activeEndpointCount"] if active_count != declared_count or active_count != EXPECTED_ACTIVE_ENDPOINT_COUNT: raise BuildInputManifestError( f"active ProductSpec count differs: actual={active_count}, " f"manifest={declared_count}, expected={EXPECTED_ACTIVE_ENDPOINT_COUNT}" ) return sources, source_digests, product_bindings def _validate_steps( document: dict[str, Any], sources: dict[str, dict[str, Any]], *, expected_steps: Sequence[str] | None, ) -> tuple[str, ...]: raw_steps = document["steps"] if not isinstance(raw_steps, list) or len(raw_steps) != EXPECTED_STEP_COUNT: raise BuildInputManifestError(f"steps must contain exactly {EXPECTED_STEP_COUNT} entries") step_order: list[str] = [] used_sources: set[str] = set() declared_exclusions: set[str] = set() for index, raw_step in enumerate(raw_steps): step = _mapping(raw_step, f"steps[{index}]") step_keys = {"name", "mutationMode", "inputs", "exclusions", "outputs"} _exact_keys(step, step_keys, f"steps[{index}]") name = str(step.get("name") or "") if not _TABLE_RE.fullmatch(name): raise BuildInputManifestError(f"steps[{index}].name is invalid: {name!r}") if name in step_order: raise BuildInputManifestError(f"duplicate step: {name}") mutation_mode = str(step.get("mutationMode") or "") if mutation_mode not in _MUTATION_MODE_POLICIES: raise BuildInputManifestError(f"step {name} has unknown mutationMode {mutation_mode!r}") inputs = _mapping(step["inputs"], f"step {name}.inputs") _exact_keys(inputs, {"sources", "derivedSteps"}, f"step {name}.inputs") input_sources = _strings(inputs["sources"], f"step {name}.inputs.sources", allow_empty=True) derived_steps = _strings( inputs["derivedSteps"], f"step {name}.inputs.derivedSteps", allow_empty=True ) if not input_sources and not derived_steps: raise BuildInputManifestError(f"step {name} has no declared input") for source_id in input_sources: source = sources.get(source_id) if source is None: raise BuildInputManifestError(f"step {name} declared input missing source: {source_id}") policy = _SOURCE_MODE_POLICIES[source["mode"]] if not policy["buildEligible"]: raise BuildInputManifestError(f"step {name} cannot consume excluded source: {source_id}") used_sources.add(source_id) for dependency in derived_steps: if dependency not in step_order: raise BuildInputManifestError(f"step {name} derived input is missing or not earlier: {dependency}") exclusions = _strings(step["exclusions"], f"step {name}.exclusions", allow_empty=True) for source_id in exclusions: source = sources.get(source_id) if source is None: raise BuildInputManifestError(f"step {name} exclusion references missing source: {source_id}") if _SOURCE_MODE_POLICIES[source["mode"]]["buildEligible"]: raise BuildInputManifestError(f"step {name} cannot exclude build-eligible source: {source_id}") declared_exclusions.add(source_id) outputs = _strings(step["outputs"], f"step {name}.outputs") if not all(_OUTPUT_RE.fullmatch(output) for output in outputs): raise BuildInputManifestError(f"step {name}.outputs contains an invalid target") step_order.append(name) if expected_steps is not None and list(expected_steps) != step_order: raise BuildInputManifestError( f"STEPS order differs: manifest={step_order}, executable={list(expected_steps)}" ) accounted = used_sources | declared_exclusions missing = set(sources) - accounted if missing: raise BuildInputManifestError(f"manifest contains unassigned sources: {sorted(missing)}") active_ids = {key for key, source in sources.items() if source["mode"] == "active_product_spec"} if not active_ids.issubset(used_sources): raise BuildInputManifestError( f"active ProductSpec sources are not consumed: {sorted(active_ids - used_sources)}" ) return tuple(step_order) def load_build_input_manifest( manifest_path: Path | str | None = None, *, product_specs_path: Path | str | None = None, repository_root: Path | str | None = None, expected_steps: Sequence[str] | None = None, ) -> BuildInputManifest: """Load and validate the policy without importing the build orchestrator.""" root = Path(repository_root).resolve() if repository_root else _repo_root() path = Path(manifest_path).resolve() if manifest_path else _default_manifest_path() products_path = ( Path(product_specs_path).resolve() if product_specs_path else _default_product_specs_path(root) ) document = _mapping(_load_yaml(path), str(path)) root_keys = {"apiVersion", "kind", "metadata", "sourceModePolicies", "mutationModePolicies", "sources", "steps"} _exact_keys(document, root_keys, "manifest root") if document["apiVersion"] != API_VERSION or document["kind"] != KIND: raise BuildInputManifestError(f"manifest must be {API_VERSION} kind={KIND}") metadata = _mapping(document["metadata"], "metadata") _exact_keys(metadata, {"name", "lifecycle", "activeEndpointCount"}, "metadata") if metadata["name"] != "kernel-build-all-inputs" or metadata["lifecycle"] != "active": raise BuildInputManifestError("manifest metadata must name an active kernel build") if metadata["activeEndpointCount"] != EXPECTED_ACTIVE_ENDPOINT_COUNT: raise BuildInputManifestError( f"metadata.activeEndpointCount must be {EXPECTED_ACTIVE_ENDPOINT_COUNT}" ) _validate_policies(document) sources, source_digests, product_bindings = _validate_sources( document, root=root, products_path=products_path ) step_order = _validate_steps(document, sources, expected_steps=expected_steps) active_endpoints = tuple(sorted( source["endpoint"] for source in sources.values() if source["mode"] == "active_product_spec" )) digest_material = { "digestSchemaVersion": _DIGEST_SCHEMA_VERSION, "manifest": document, "productBindings": product_bindings, "sourceDigests": source_digests, } serialized = json.dumps( digest_material, ensure_ascii=False, sort_keys=True, separators=(",", ":") ).encode("utf-8") policy_digest = hashlib.sha256(serialized).hexdigest() return BuildInputManifest( path=path, document=_deep_freeze(document), step_order=step_order, active_endpoints=active_endpoints, policy_digest=policy_digest, source_digests=MappingProxyType(dict(sorted(source_digests.items()))), product_bindings=_deep_freeze(dict(sorted(product_bindings.items()))), ) def load_product_only_input_manifest( manifest_path: Path | str, *, expected_name: str, expected_endpoint: str, expected_consumer: str, expected_mutation_mode: str, expected_outputs: Sequence[str], product_specs_path: Path | str | None = None, repository_root: Path | str | None = None, ) -> BuildInputManifest: """Load one isolated ProductSpec -> kernel projection contract. Product-only projections (currently Golden Apple) deliberately do not join the 15-step ``build_all`` graph or its 1C reference snapshot. They still reuse the exact ProductSpec validation and digest binding used by the main manifest. The returned shape implements the same immutable resolver interface, while its policy digest covers this smaller manifest, the compiled ProductSpec binding and the ProductSpec file digest. """ root = Path(repository_root).resolve() if repository_root else _repo_root() path = Path(manifest_path).resolve() products_path = ( Path(product_specs_path).resolve() if product_specs_path else _default_product_specs_path(root) ) document = _mapping(_load_yaml(path), str(path)) _exact_keys( document, {"apiVersion", "kind", "metadata", "source", "consumer"}, "product-only manifest root", ) if document["apiVersion"] != API_VERSION or document["kind"] != PRODUCT_ONLY_KIND: raise BuildInputManifestError( f"product-only manifest must be {API_VERSION} kind={PRODUCT_ONLY_KIND}" ) metadata = _mapping(document["metadata"], "metadata") _exact_keys(metadata, {"name", "lifecycle", "activeEndpointCount"}, "metadata") if metadata != { "name": expected_name, "lifecycle": "active", "activeEndpointCount": 1, }: raise BuildInputManifestError( "product-only metadata differs from its exact active contract" ) source = _mapping(document["source"], "source") if source.get("mode") != "active_product_spec": raise BuildInputManifestError( "product-only source must use active_product_spec mode" ) raw = _mapping( _load_yaml(products_path / f"{source.get('endpoint')}.yaml"), f"ProductSpec {source.get('endpoint')}", ) try: binding = validate_product_input_source( source, mode="active_product_spec", products_path=products_path, raw=raw, ) except ProductInputContractError as exc: raise BuildInputManifestError(str(exc)) from exc if binding["endpoint"] != expected_endpoint: raise BuildInputManifestError( f"product-only endpoint differs: {binding['endpoint']!r} != {expected_endpoint!r}" ) if source.get("id") != f"product:{expected_endpoint}": raise BuildInputManifestError( "product-only source id must be the exact product endpoint id" ) consumer = _mapping(document["consumer"], "consumer") _exact_keys(consumer, {"module", "mutationMode", "outputs"}, "consumer") if consumer["module"] != expected_consumer: raise BuildInputManifestError( f"product-only consumer differs: {consumer['module']!r} != {expected_consumer!r}" ) if ( expected_mutation_mode not in _MUTATION_MODE_POLICIES or consumer["mutationMode"] != expected_mutation_mode ): raise BuildInputManifestError( "product-only consumer mutationMode differs from its exact contract" ) outputs = _strings(consumer["outputs"], "consumer.outputs") if not all(_OUTPUT_RE.fullmatch(output) for output in outputs): raise BuildInputManifestError("consumer.outputs contains an invalid target") if tuple(outputs) != tuple(expected_outputs): raise BuildInputManifestError( "product-only consumer outputs differ from their exact contract" ) source_digest = binding["digest"] resolver_document = { **document, # The shared resolver intentionally consumes a normalized sources list. "sources": [source], } digest_material = { "digestSchemaVersion": _DIGEST_SCHEMA_VERSION, "manifest": document, "productBindings": {expected_endpoint: binding}, "sourceDigests": {source["id"]: source_digest}, } serialized = json.dumps( digest_material, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return BuildInputManifest( path=path, document=_deep_freeze(resolver_document), step_order=(expected_consumer,), active_endpoints=(expected_endpoint,), policy_digest=hashlib.sha256(serialized).hexdigest(), source_digests=MappingProxyType({source["id"]: source_digest}), product_bindings=_deep_freeze({expected_endpoint: binding}), ) def validate_build_input_manifest( manifest_path: Path | str | None = None, **kwargs: Any, ) -> BuildInputManifest: """Validate the policy against ``build_all.STEPS`` without importing DB code.""" build_all_path = Path(__file__).with_name("build_all.py") try: tree = ast.parse(build_all_path.read_text(encoding="utf-8"), build_all_path.name) except (OSError, SyntaxError) as exc: raise BuildInputManifestError(f"cannot inspect build_all.STEPS: {exc}") from exc executable_order: list[str] | None = None for node in tree.body: if not isinstance(node, ast.Assign): continue if not any(isinstance(target, ast.Name) and target.id == "STEPS" for target in node.targets): continue if not isinstance(node.value, (ast.List, ast.Tuple)): break names: list[str] = [] for item in node.value.elts: if ( not isinstance(item, ast.Tuple) or len(item.elts) != 2 or not isinstance(item.elts[0], ast.Constant) or not isinstance(item.elts[0].value, str) ): names = [] break names.append(item.elts[0].value) if names: executable_order = names break if executable_order is None: raise BuildInputManifestError("cannot derive literal build_all.STEPS order") if "expected_steps" in kwargs: raise TypeError("validate_build_input_manifest derives expected_steps from build_all") manifest = load_build_input_manifest( manifest_path, expected_steps=executable_order, **kwargs, ) # Local import avoids a module cycle: the source scanner uses the validated # manifest types, while this public checker must also enforce code closure. from kernel.etl.build_input_source_closure import ( # noqa: PLC0415 validate_executable_source_closure, ) validate_executable_source_closure(manifest) return manifest