"""Materialize the Core-owned GenerationPlan composition into a repo. The provider may edit customer-facing files, but it must not be responsible for declaring the architecture selected by Core. These files are deliberately written before provider execution and are treated as protected control-plane artifacts by repository export. """ from __future__ import annotations import json from pathlib import Path from typing import Any from src.domain.app_blocks.catalogs import canonical_feature_pack from src.domain.app_blocks.composition_graph import ( selected_composition_graph, validate_composition_graph, ) from src.domain.app_blocks.hosts import host_map_payload from src.domain.app_blocks.registry import load_default_registry from src.domain.app_blocks.resolver import planned_from_manifest from src.domain.app_blocks.schemas import AppBlockPlan from src.domain.starter_library.catalog import contract_map_payload def _dependency_closure( *, block_ids: list[str], explicit_pack_ids: set[str], ) -> tuple[set[str], set[str]]: """Resolve App Block and catalog-pack dependencies into canonical ids. GenerationPlan is the authoritative composition input, but older plans may omit the redundant pack projection. Resolve that projection here so every consumer (generated manifests and operator repair) sees the same closure. """ from src.domain.feature_packs.normalize import load_pack_contract registry = load_default_registry() packs = {canonical_feature_pack(item) for item in explicit_pack_ids} primitives: set[str] = set() visited_blocks: set[str] = set() visited_packs: set[str] = set() def visit_block(block_id: str) -> None: if block_id in visited_blocks: return visited_blocks.add(block_id) manifest = registry.get(block_id) if manifest is None: raise ValueError(f"generation_plan_unknown_app_block:{block_id}") for dep in manifest.requires: if dep.kind == "feature_pack": visit_pack(dep.id) elif dep.kind == "primitive": primitives.add(dep.id) elif dep.kind == "app_block": visit_block(dep.id) if manifest.composition_fallback and manifest.composition_fallback.supported: for nested in manifest.composition_fallback.blocks: visit_block(nested) for pack in manifest.composition_fallback.feature_packs: visit_pack(pack) def visit_pack(pack_id: str) -> None: slug = canonical_feature_pack(pack_id) packs.add(slug) if slug in visited_packs: return visited_packs.add(slug) # Catalog overlays are the hosting-neutral pack dependency contract. # Unknown/project-local packs remain explicit and are validated later # by the installer rather than being silently discarded. try: contract = load_pack_contract(slug, {}) except (KeyError, TypeError, ValueError): return for dep in contract.requires: if dep.kind == "feature_pack": visit_pack(dep.id) elif dep.kind == "primitive": primitives.add(dep.id) for block_id in sorted(set(block_ids)): visit_block(block_id) for pack_id in sorted(set(explicit_pack_ids)): visit_pack(pack_id) return packs, primitives def _core_app_block_plan(generation_plan: dict[str, Any]) -> dict[str, Any]: composition = generation_plan.get("composition") or {} block_ids = [str(item) for item in composition.get("app_block_ids") or []] explicit_pack_ids = {str(item) for item in composition.get("feature_pack_ids") or []} primitive_ids = {str(item) for item in composition.get("primitive_ids") or []} cloud_ids = [str(item) for item in composition.get("cloud_capability_ids") or []] registry = load_default_registry() blocks = [] for block_id in sorted(set(block_ids)): manifest = registry.get(block_id) if manifest is None: raise ValueError(f"generation_plan_unknown_app_block:{block_id}") blocks.append( planned_from_manifest( manifest, "Required by Core GenerationPlan.", {}, ).model_dump(mode="json", exclude={"optional_requirements"}) ) pack_ids, derived_primitives = _dependency_closure( block_ids=block_ids, explicit_pack_ids=explicit_pack_ids, ) primitive_ids.update(derived_primitives) runtime = generation_plan.get("runtime") or {} payload = AppBlockPlan( requested_capabilities=sorted(set(block_ids + cloud_ids)), blocks=blocks, required_feature_packs=sorted(pack_ids), required_primitives=sorted(primitive_ids), runtime_dependencies=sorted(set(runtime.get("runtime_dependencies") or [])), starter_contracts=sorted(set(runtime.get("starter_contract_ids") or [])), adoption="authoritative", feature_pack_ids=sorted(pack_ids), ) return payload.model_dump(mode="json") def _resolved_form_definitions(generation_plan: dict[str, Any]) -> dict[str, dict[str, Any]]: """Return the typed form projection carried by the completed plan. This is intentionally a projection only. The GenerationPlan has already validated the schema; this layer must not infer fields from generated markup. """ raw = ((generation_plan.get("composition") or {}).get("form_definitions") or {}) if not isinstance(raw, dict): return {} return { str(key): value for key, value in raw.items() if ( isinstance(value, dict) and value.get("key") == key and isinstance(value.get("fields"), list) ) } def render_core_composition_files(generation_plan: dict[str, Any]) -> dict[str, str]: """Return protected composition files for a provider workspace. ``app-recipe.json`` and the Cloud port contract are installed by the platform-adapter phase. This function owns the remaining composition projections and never invents Cloud routes or schemas from capability IDs. """ composition = generation_plan.get("composition") or {} core_plan = _core_app_block_plan(generation_plan) packs = list(core_plan["required_feature_packs"]) composition_graph = selected_composition_graph( app_block_ids=[str(item) for item in composition.get("app_block_ids") or []], feature_pack_ids=[str(item) for item in composition.get("feature_pack_ids") or []], ) composition_graph_errors = validate_composition_graph(composition_graph) core_composition = { "schema_version": "core_composition.v1", "provenance": generation_plan.get("provenance") or {}, "composition": composition, "runtime": generation_plan.get("runtime") or {}, "provider_requirements": generation_plan.get("provider_requirements") or [], "constraints": generation_plan.get("constraints") or {}, "validation": generation_plan.get("validation") or {}, "gaps": generation_plan.get("gaps") or {}, "composition_graph": composition_graph, "composition_graph_validation": { "valid": not composition_graph_errors, "errors": composition_graph_errors, }, } unsupported = (generation_plan.get("gaps") or {}).get("unsupported_semantics") or [] project_id = ((generation_plan.get("metadata") or {}).get("project_id")) l3_candidates = [] for gap in unsupported: if not isinstance(gap, dict): continue semantic = str(gap.get("id") or "").strip() if not semantic: continue candidate_id = f"l3.{semantic}@project:{project_id or 'pending'}" l3_candidates.append( { "candidate_id": candidate_id, "semantic": semantic, "scope": "project", "status": "discovered", "requirement_summary": str(gap.get("message") or semantic), "gap_reason": str(gap.get("kind") or "semantic_gap"), "dependencies": list(gap.get("affects") or []), "architecture_auto_promotion": False, "implementation_boundary": ( "project-scoped generated code using admitted P2HS Cloud routes; " "no Core catalog mutation and no private backend" ), } ) form_definitions = _resolved_form_definitions(generation_plan) install = [] for slug in packs: config: dict[str, Any] = {} if slug == "contact-capture" and form_definitions: config["form_definitions"] = form_definitions install.append({"pack_slug": slug, "config": config}) return { "p2hs/generation-plan.json": json.dumps( generation_plan, indent=2 ) + "\n", "p2hs/app-blocks-plan.json": json.dumps( core_plan, indent=2 ) + "\n", "p2hs/feature-packs.json": json.dumps( { "version": 1, "install": install, }, indent=2, ) + "\n", "p2hs/starter-contracts.json": json.dumps( contract_map_payload(), indent=2 ) + "\n", "p2hs/app-block-hosts.json": json.dumps(host_map_payload(), indent=2) + "\n", "p2hs/core-composition.json": json.dumps(core_composition, indent=2) + "\n", "p2hs/l3-plan.json": json.dumps( { "schema_version": "l3_plan.v1", "architecture_auto_promotion": False, "candidates": l3_candidates, }, indent=2, ) + "\n", } def materialize_core_composition(repo_path: str | Path, generation_plan: dict[str, Any]) -> None: """Write the exact Core-owned composition projection. ``feature-packs.json`` is a control-plane manifest, not an AI-owned extension point. Keeping old ``install`` or ``compose`` entries here lets a provider smuggle stale or unplanned capabilities into the provisioning gate. Any additional capability must first be admitted into the Core GenerationPlan. """ root = Path(repo_path) files = render_core_composition_files(generation_plan) p2hs = root / "p2hs" p2hs.mkdir(parents=True, exist_ok=True) (p2hs / "app-blocks-plan.json").write_text( files["p2hs/app-blocks-plan.json"], encoding="utf-8" ) (p2hs / "generation-plan.json").write_text( files["p2hs/generation-plan.json"], encoding="utf-8" ) from src.domain.codegen.application_experience import published_experience (p2hs / "application-experience.json").write_text( json.dumps(published_experience(generation_plan.get("experience")), indent=2) + "\n", encoding="utf-8", ) (p2hs / "core-composition.json").write_text( files["p2hs/core-composition.json"], encoding="utf-8" ) (p2hs / "l3-plan.json").write_text( files["p2hs/l3-plan.json"], encoding="utf-8" ) try: from src.domain.app_blocks.hosts import merge_host_map existing_hosts = json.loads((p2hs / "app-block-hosts.json").read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError): existing_hosts = None (p2hs / "app-block-hosts.json").write_text( json.dumps( merge_host_map(existing_hosts, json.loads(files["p2hs/app-block-hosts.json"])), indent=2, ) + "\n", encoding="utf-8", ) try: from src.domain.starter_library.catalog import merge_contract_map existing_contracts = json.loads( (p2hs / "starter-contracts.json").read_text(encoding="utf-8") ) except (OSError, UnicodeDecodeError, json.JSONDecodeError): existing_contracts = None (p2hs / "starter-contracts.json").write_text( json.dumps( merge_contract_map( existing_contracts, json.loads(files["p2hs/starter-contracts.json"]), ), indent=2, ) + "\n", encoding="utf-8", ) # Replace the whole file deliberately. In particular, do not preserve # provider-declared ``compose`` entries or stale installs from a previous # attempt: Core owns this manifest and the runner validates this exact # projection before it asks Cloud to provision anything. (p2hs / "feature-packs.json").write_text( files["p2hs/feature-packs.json"], encoding="utf-8" ) __all__ = ["materialize_core_composition", "render_core_composition_files"]