"""Derived Pack -> App Block -> Primitive composition graph. The authored sources remain the App Block YAML manifests and Feature Pack contracts. This module only projects those sources into one deterministic, read-only graph for Codegen, audits, and coverage tooling. It deliberately does not create dependencies, infer missing primitives, or change install behavior. A missing edge is evidence for the catalog owners, not an invitation to synthesize a backend capability. """ from __future__ import annotations from typing import Any from src.domain.app_blocks.canonical import CANONICAL_REQUIREMENT_CAPABILITIES from src.domain.app_blocks.catalogs import default_known_packs from src.domain.app_blocks.registry import load_default_registry from src.domain.feature_packs.catalog_v2 import CATALOG_OVERLAYS from src.domain.feature_packs.normalize import load_pack_contract COMPOSITION_GRAPH_SCHEMA_VERSION = 1 def _dependency_rows(dependencies: list[Any]) -> list[dict[str, str]]: return [ {"kind": dependency.kind, "id": dependency.id} for dependency in dependencies ] def build_composition_graph() -> dict[str, Any]: """Return the canonical, deterministic composition graph. Pack ids include both the default Core catalog and V2 overlay catalog so the graph remains useful while persisted pack rows are being migrated. """ registry = load_default_registry() pack_ids = sorted(set(default_known_packs()) | set(CATALOG_OVERLAYS)) blocks: list[dict[str, Any]] = [] for canonical in sorted(registry.canonical.blocks, key=lambda item: item.id): manifest = canonical.manifest blocks.append( { "id": canonical.id, "version": canonical.version, "family": canonical.family, "title": canonical.title, "status": manifest.status, "feature_packs": list(canonical.installation.feature_packs), "requires": _dependency_rows(list(canonical.requires)), "optional_requires": _dependency_rows(list(canonical.optional_requires)), "provides": [item.id for item in manifest.provides], } ) packs: list[dict[str, Any]] = [] block_by_pack = { pack_id: list(registry.canonical.blocks_for_feature_pack(pack_id)) for pack_id in pack_ids } for pack_id in pack_ids: contract = load_pack_contract(pack_id, {}) packs.append( { "id": pack_id, "app_blocks": block_by_pack[pack_id], "provides": contract.provide_ids(), "requires": _dependency_rows(contract.requires), "optional_requires": _dependency_rows(contract.optional_requires), } ) return { "schema_version": COMPOSITION_GRAPH_SCHEMA_VERSION, "authority": { "app_blocks": "core_yaml", "feature_packs": "core_catalog_and_persisted_contract", "primitives": "feature_pack_contract", }, "packs": packs, "app_blocks": blocks, } def composition_graph_payload() -> dict[str, Any]: """Stable public alias for callers exporting the graph.""" return build_composition_graph() def selected_composition_graph( *, app_block_ids: list[str], feature_pack_ids: list[str] ) -> dict[str, Any]: """Project the catalog graph onto one Core-selected composition. Unknown ids are retained in ``unresolved`` so an audit can report a missing edge without silently treating it as satisfied. """ graph = build_composition_graph() requested_block_ids = set(app_block_ids) requested_pack_ids = set(feature_pack_ids) blocks_by_id = {item["id"]: item for item in graph["app_blocks"]} packs_by_id = {item["id"]: item for item in graph["packs"]} pending = list(requested_block_ids) while pending: block_id = pending.pop() block = blocks_by_id.get(block_id) if block is None: continue for dependency in block.get("requires", []): if dependency.get("kind") != "app_block": continue dependency_id = dependency.get("id") if dependency_id not in requested_block_ids: requested_block_ids.add(dependency_id) pending.append(dependency_id) requested_blocks = sorted(requested_block_ids) for block_id in requested_blocks: block = blocks_by_id.get(block_id) if block is not None: requested_pack_ids.update(block.get("feature_packs", [])) pending_packs = list(requested_pack_ids) while pending_packs: pack_id = pending_packs.pop() pack = packs_by_id.get(pack_id) if pack is None: continue for dependency in pack.get("requires", []): if dependency.get("kind") != "feature_pack": continue dependency_id = dependency.get("id") if dependency_id not in requested_pack_ids: requested_pack_ids.add(dependency_id) pending_packs.append(dependency_id) requested_packs = sorted(requested_pack_ids) selected_packs = [] for item in requested_packs: pack = packs_by_id.get(item) if pack is None: continue selected_packs.append( { **pack, "app_blocks": [ block_id for block_id in pack.get("app_blocks", []) if block_id in requested_blocks ], } ) return { **graph, "scope": "selected_composition", "packs": selected_packs, "app_blocks": [ blocks_by_id[item] for item in requested_blocks if item in blocks_by_id ], "unresolved": { "feature_packs": [item for item in requested_packs if item not in packs_by_id], "app_blocks": [item for item in requested_blocks if item not in blocks_by_id], }, } def validate_composition_graph(graph: dict[str, Any]) -> list[str]: """Return deterministic architecture errors for a graph payload.""" blocks = {item["id"]: item for item in graph.get("app_blocks") or []} packs = {item["id"]: item for item in graph.get("packs") or []} errors: set[str] = set() for block in blocks.values(): block_id = block["id"] for dependency in block.get("requires", []): kind = dependency.get("kind") dependency_id = dependency.get("id") if ( kind == "primitive" and dependency_id not in CANONICAL_REQUIREMENT_CAPABILITIES ): errors.add(f"unknown_primitive:{block_id}:{dependency_id}") elif kind == "feature_pack" and dependency_id not in packs: errors.add(f"unknown_feature_pack:{block_id}:{dependency_id}") elif kind == "app_block" and dependency_id not in blocks: errors.add(f"unknown_app_block:{block_id}:{dependency_id}") for pack_id in block.get("feature_packs", []): if pack_id not in packs: errors.add(f"unknown_feature_pack:{block_id}:{pack_id}") elif block_id not in packs[pack_id].get("app_blocks", []): errors.add(f"broken_pack_block_edge:{pack_id}:{block_id}") for pack in packs.values(): pack_id = pack["id"] for block_id in pack.get("app_blocks", []): if block_id not in blocks: errors.add(f"unknown_app_block:{pack_id}:{block_id}") elif pack_id not in blocks[block_id].get("feature_packs", []): errors.add(f"broken_pack_block_edge:{pack_id}:{block_id}") for dependency in pack.get("requires", []) + pack.get("optional_requires", []): if ( dependency.get("kind") == "primitive" and dependency.get("id") not in CANONICAL_REQUIREMENT_CAPABILITIES ): errors.add(f"unknown_primitive:{pack_id}:{dependency.get('id')}") unresolved = graph.get("unresolved") or {} for kind in ("feature_packs", "app_blocks"): for item in unresolved.get(kind) or []: errors.add(f"unresolved_{kind}:{item}") return sorted(errors)